


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







     def 							lowercase__       (						self :						List[str]      ):
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		'''| <pad> <unk> <s> </s> a b c d e f g h i j k'''.split()
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		dict(zip(__UpperCAmelCase   ,   range(len(__UpperCAmelCase      )      )      )      )

           SCREAMING_SNAKE_CASE__       :  Dict						     =		{
               '''unk_token''': '''<unk>''',
               '''bos_token''': '''<s>''',
               '''eos_token''': '''</s>''',
           }
           SCREAMING_SNAKE_CASE__       :  int						     =		{
               '''feature_size''': 1,
               '''padding_value''': 0.0,
               '''sampling_rate''': 1_60_00,
               '''return_attention_mask''': False,
               '''do_normalize''': True,
           }

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

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

           # load decoder from hub
           SCREAMING_SNAKE_CASE__       :  int						     =		'''hf-internal-testing/ngram-beam-search-decoder'''







     def 							lowercase__       (						self :						str   ,   **_lowercase :						Optional[int]      ):
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		self.add_kwargs_tokens_map.copy()
           kwargs.update(__UpperCAmelCase      )
           return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname   ,   **__UpperCAmelCase      )







     def 							lowercase__       (						self :						List[Any]   ,   **_lowercase :						Tuple      ):
           return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname   ,   **__UpperCAmelCase      )







     def 							lowercase__       (						self :						str   ,   **_lowercase :						Dict      ):
           return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name   ,   **__UpperCAmelCase      )







     def 							lowercase__       (						self :						int      ):
           shutil.rmtree(self.tmpdirname      )







     def 							lowercase__       (						self :						List[Any]      ):
           SCREAMING_SNAKE_CASE__       :  str						     =		self.get_tokenizer()
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		self.get_feature_extractor()
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.get_decoder()

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

           processor.save_pretrained(self.tmpdirname      )
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname      )

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

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

           # 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   ,   __UpperCAmelCase      )







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







     def 							lowercase__       (						self :						str      ):
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		self.get_feature_extractor()
           SCREAMING_SNAKE_CASE__       :  str						     =		self.get_tokenizer()
           SCREAMING_SNAKE_CASE__       :  str						     =		self.get_decoder()

           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		WavaVecaProcessorWithLM(tokenizer=__UpperCAmelCase   ,   feature_extractor=__UpperCAmelCase   ,   decoder=__UpperCAmelCase      )

           SCREAMING_SNAKE_CASE__       :  Dict						     =		floats_list((3, 10_00)      )

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

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

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

           SCREAMING_SNAKE_CASE__       :  str						     =		processor(text=__UpperCAmelCase      )

           SCREAMING_SNAKE_CASE__       :  Tuple						     =		tokenizer(__UpperCAmelCase      )

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







     def 							lowercase__       (						self :						int   ,   _lowercase :						Dict=(2, 10, 16)   ,   _lowercase :						Tuple=77      ):
           np.random.seed(__UpperCAmelCase      )
           return np.random.rand(*__UpperCAmelCase      )







     def 							lowercase__       (						self :						int      ):
           SCREAMING_SNAKE_CASE__       :  Any						     =		self.get_feature_extractor()
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		self.get_tokenizer()
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		self.get_decoder()

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

           SCREAMING_SNAKE_CASE__       :  Dict						     =		self._get_dummy_logits(shape=(10, 16)   ,   seed=13      )

           SCREAMING_SNAKE_CASE__       :  List[str]						     =		processor.decode(__UpperCAmelCase      )

           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		decoder.decode_beams(__UpperCAmelCase      )[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 							lowercase__       (						self :						Dict   ,   _lowercase :						Union[str, Any]      ):
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		self.get_feature_extractor()
           SCREAMING_SNAKE_CASE__       :  Dict						     =		self.get_tokenizer()
           SCREAMING_SNAKE_CASE__       :  Dict						     =		self.get_decoder()

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

           SCREAMING_SNAKE_CASE__       :  int						     =		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__       :  Any						     =		processor.batch_decode(__UpperCAmelCase      )
           else:
                 with get_context(__UpperCAmelCase      ).Pool() as pool:
                       SCREAMING_SNAKE_CASE__       :  Dict						     =		processor.batch_decode(__UpperCAmelCase   ,   __UpperCAmelCase      )

           SCREAMING_SNAKE_CASE__       :  str						     =		list(__UpperCAmelCase      )

           with get_context('''fork'''      ).Pool() as p:
                 SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		decoder.decode_beams_batch(__UpperCAmelCase   ,   __UpperCAmelCase      )

           SCREAMING_SNAKE_CASE__     ,       SCREAMING_SNAKE_CASE__     ,       SCREAMING_SNAKE_CASE__       :  Tuple						     =		[], [], []
           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(__UpperCAmelCase   ,   decoded_processor.text      )
           self.assertListEqual(['''<s> <s> </s>''', '''<s> <s> <s>''']   ,   decoded_processor.text      )
           self.assertListEqual(__UpperCAmelCase   ,   decoded_processor.logit_score      )
           self.assertListEqual(__UpperCAmelCase   ,   decoded_processor.lm_score      )







     def 							lowercase__       (						self :						List[Any]      ):
           SCREAMING_SNAKE_CASE__       :  Dict						     =		self.get_feature_extractor()
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		self.get_tokenizer()
           SCREAMING_SNAKE_CASE__       :  str						     =		self.get_decoder()

           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		WavaVecaProcessorWithLM(tokenizer=__UpperCAmelCase   ,   feature_extractor=__UpperCAmelCase   ,   decoder=__UpperCAmelCase      )

           SCREAMING_SNAKE_CASE__       :  Any						     =		self._get_dummy_logits()

           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		15
           SCREAMING_SNAKE_CASE__       :  Dict						     =		-20.0
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		-4.0

           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		processor.batch_decode(
               __UpperCAmelCase   ,   beam_width=__UpperCAmelCase   ,   beam_prune_logp=__UpperCAmelCase   ,   token_min_logp=__UpperCAmelCase   ,   )
           SCREAMING_SNAKE_CASE__       :  Dict						     =		decoded_processor_out.text

           SCREAMING_SNAKE_CASE__       :  Dict						     =		list(__UpperCAmelCase      )

           with get_context('''fork'''      ).Pool() as pool:
                 SCREAMING_SNAKE_CASE__       :  int						     =		decoder.decode_beams_batch(
                     __UpperCAmelCase   ,   __UpperCAmelCase   ,   beam_width=__UpperCAmelCase   ,   beam_prune_logp=__UpperCAmelCase   ,   token_min_logp=__UpperCAmelCase   ,   )

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

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

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

           self.assertTrue(np.array_equal(__UpperCAmelCase   ,   decoded_processor_out.lm_score      )      )
           self.assertTrue(np.allclose([-15.554, -13.9474]   ,   __UpperCAmelCase   ,   atol=1E-3      )      )







     def 							lowercase__       (						self :						int      ):
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		self.get_feature_extractor()
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		self.get_tokenizer()
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		self.get_decoder()

           SCREAMING_SNAKE_CASE__       :  str						     =		WavaVecaProcessorWithLM(tokenizer=__UpperCAmelCase   ,   feature_extractor=__UpperCAmelCase   ,   decoder=__UpperCAmelCase      )

           SCREAMING_SNAKE_CASE__       :  Tuple						     =		self._get_dummy_logits()

           SCREAMING_SNAKE_CASE__       :  int						     =		2.0
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		5.0
           SCREAMING_SNAKE_CASE__       :  Dict						     =		-20.0
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		True

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

           SCREAMING_SNAKE_CASE__       :  Dict						     =		list(__UpperCAmelCase      )
           decoder.reset_params(
               alpha=__UpperCAmelCase   ,   beta=__UpperCAmelCase   ,   unk_score_offset=__UpperCAmelCase   ,   lm_score_boundary=__UpperCAmelCase   ,   )

           with get_context('''fork'''      ).Pool() as pool:
                 SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		decoder.decode_beams_batch(
                     __UpperCAmelCase   ,   __UpperCAmelCase   ,   )

           SCREAMING_SNAKE_CASE__       :  List[str]						     =		[d[0][0] for d in decoded_decoder_out]

           self.assertListEqual(__UpperCAmelCase   ,   __UpperCAmelCase      )
           self.assertListEqual(['''<s> </s> <s> </s> </s>''', '''</s> </s> <s> </s> </s>''']   ,   __UpperCAmelCase      )
           SCREAMING_SNAKE_CASE__       :  Union[str, 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   ,   __UpperCAmelCase      )







     def 							lowercase__       (						self :						str      ):
           SCREAMING_SNAKE_CASE__       :  List[str]						     =		WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )

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

           SCREAMING_SNAKE_CASE__       :  Any						     =		os.listdir(__UpperCAmelCase      )
           SCREAMING_SNAKE_CASE__       :  Dict						     =		['''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(__UpperCAmelCase   ,   __UpperCAmelCase      )







     def 							lowercase__       (						self :						Optional[Any]      ):
           SCREAMING_SNAKE_CASE__       :  Any						     =		snapshot_download('''hf-internal-testing/processor_with_lm'''      )

           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		WavaVecaProcessorWithLM.from_pretrained(__UpperCAmelCase      )

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

           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		os.listdir(__UpperCAmelCase      )
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		os.listdir(__UpperCAmelCase      )

           local_decoder_files.sort()
           expected_decoder_files.sort()

           # test that both decoder form hub and local files in cache are the same
           self.assertListEqual(__UpperCAmelCase   ,   __UpperCAmelCase      )







     def 							lowercase__       (						self :						Dict      ):
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''      )

           SCREAMING_SNAKE_CASE__       :  Tuple						     =		floats_list((3, 10_00)      )

           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		processor_wavaveca(__UpperCAmelCase   ,   return_tensors='''np'''      )
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		processor_auto(__UpperCAmelCase   ,   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(__UpperCAmelCase      )
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		processor_auto.batch_decode(__UpperCAmelCase      )

           self.assertListEqual(decoded_wavaveca.text   ,   decoded_auto.text      )







     def 							lowercase__       (						self :						Optional[int]      ):
           SCREAMING_SNAKE_CASE__       :  str						     =		self.get_feature_extractor()
           SCREAMING_SNAKE_CASE__       :  Any						     =		self.get_tokenizer()
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.get_decoder()

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

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







     @staticmethod
     def 							lowercase__       (						_lowercase :						int   ,   _lowercase :						List[str]      ):
           SCREAMING_SNAKE_CASE__       :  str						     =		[d[key] for d in offsets]
           return retrieved_list







     def 							lowercase__       (						self :						Optional[Any]      ):
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		self._get_dummy_logits()[0]

           SCREAMING_SNAKE_CASE__       :  Any						     =		processor.decode(__UpperCAmelCase   ,   output_word_offsets=__UpperCAmelCase      )
           # 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(__UpperCAmelCase   ,   __UpperCAmelCase      )      )

           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 							lowercase__       (						self :						List[str]      ):
           SCREAMING_SNAKE_CASE__       :  Dict						     =		WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		self._get_dummy_logits()

           SCREAMING_SNAKE_CASE__       :  List[str]						     =		processor.batch_decode(__UpperCAmelCase   ,   output_word_offsets=__UpperCAmelCase      )

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

           self.assertListEqual(
               [''' '''.join(self.get_from_offsets(__UpperCAmelCase   ,   '''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 							lowercase__       (						self :						List[str]      ):
           import torch

           SCREAMING_SNAKE_CASE__       :  Any						     =		load_dataset('''common_voice'''   ,   '''en'''   ,   split='''train'''   ,   streaming=__UpperCAmelCase      )
           SCREAMING_SNAKE_CASE__       :  str						     =		ds.cast_column('''audio'''   ,   datasets.Audio(sampling_rate=1_60_00      )      )
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		iter(__UpperCAmelCase      )
           SCREAMING_SNAKE_CASE__       :  str						     =		next(__UpperCAmelCase      )

           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__       :  List[Any]						     =		processor(sample['''audio''']['''array''']   ,   return_tensors='''pt'''      ).input_values

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

           SCREAMING_SNAKE_CASE__       :  Dict						     =		processor.decode(logits[0]   ,   output_word_offsets=__UpperCAmelCase      )

           SCREAMING_SNAKE_CASE__       :  str						     =		model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
           SCREAMING_SNAKE_CASE__       :  List[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__       :  Tuple						     =		'''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(__UpperCAmelCase   ,   '''word'''      )      )   ,   __UpperCAmelCase      )
           self.assertEqual(''' '''.join(self.get_from_offsets(__UpperCAmelCase   ,   '''word'''      )      )   ,   output.text      )

           # output times
           SCREAMING_SNAKE_CASE__       :  int						     =		torch.tensor(self.get_from_offsets(__UpperCAmelCase   ,   '''start_time'''      )      )
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		torch.tensor(self.get_from_offsets(__UpperCAmelCase   ,   '''end_time'''      )      )

           # fmt: off
           SCREAMING_SNAKE_CASE__       :  int						     =		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__       :  Tuple						     =		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(__UpperCAmelCase   ,   __UpperCAmelCase   ,   atol=0.01      )      )
           self.assertTrue(torch.allclose(__UpperCAmelCase   ,   __UpperCAmelCase   ,   atol=0.01      )      )














from ... import PretrainedConfig


UpperCAmelCase__							  =						{
    "sijunhe/nezha-cn-base": "https://huggingface.co/sijunhe/nezha-cn-base/resolve/main/config.json",
}






class 						lowercase_   (      lowercase  ):






      '''simple docstring'''




      __snake_case	    =    NEZHA_PRETRAINED_CONFIG_ARCHIVE_MAP
      __snake_case	    =    '''nezha'''







      def __init__( self :		Union[str, Any]					,   __UpperCAmelCase :		List[str]=21_128					,   __UpperCAmelCase :		Any=768					,   __UpperCAmelCase :		int=12					,   __UpperCAmelCase :		Optional[int]=12					,   __UpperCAmelCase :		Tuple=3_072					,   __UpperCAmelCase :		Union[str, Any]="gelu"					,   __UpperCAmelCase :		List[str]=0.1					,   __UpperCAmelCase :		Any=0.1					,   __UpperCAmelCase :		str=512					,   __UpperCAmelCase :		List[Any]=64					,   __UpperCAmelCase :		Tuple=2					,   __UpperCAmelCase :		List[Any]=0.02					,   __UpperCAmelCase :		List[str]=1e-1_2					,   __UpperCAmelCase :		int=0.1					,   __UpperCAmelCase :		Dict=0					,   __UpperCAmelCase :		Any=2					,   __UpperCAmelCase :		Optional[Any]=3					,   __UpperCAmelCase :		int=True					,   **__UpperCAmelCase :		Optional[Any]					,   )       ->Any:




       """simple docstring"""


       super().__init__(pad_token_id=__UpperCAmelCase					,   bos_token_id=__UpperCAmelCase					,   eos_token_id=__UpperCAmelCase					,   **__UpperCAmelCase    )

       a						    = vocab_size
       a						    = hidden_size
       a						    = num_hidden_layers
       a						    = num_attention_heads
       a						    = hidden_act
       a						    = intermediate_size
       a						    = hidden_dropout_prob
       a						    = attention_probs_dropout_prob
       a						    = max_position_embeddings
       a						    = max_relative_position
       a						    = type_vocab_size
       a						    = initializer_range
       a						    = layer_norm_eps
       a						    = classifier_dropout
       a						    = use_cache


