
'''simple docstring'''





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    ):




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

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

										snake_case       :     List[Any] =    tempfile.mkdtemp()
										snake_case       :     List[Any] =    os.path.join(self.tmpdirname,    VOCAB_FILES_NAMES['''vocab_file''']      )
										snake_case       :     List[str] =    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
										snake_case       :     Tuple =    '''hf-internal-testing/ngram-beam-search-decoder'''




				def 	__snake_case							(							self	:				Dict,    **SCREAMING_SNAKE_CASE_	:				Union[str, Any]      ):
										snake_case       :     Union[str, Any] =    self.add_kwargs_tokens_map.copy()
										kwargs.update(SCREAMING_SNAKE_CASE_      )
										return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname,    **SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Tuple,    **SCREAMING_SNAKE_CASE_	:				str      ):
										return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname,    **SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				List[Any],    **SCREAMING_SNAKE_CASE_	:				List[str]      ):
										return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name,    **SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				List[str]      ):
										shutil.rmtree(self.tmpdirname      )




				def 	__snake_case							(							self	:				Optional[int]      ):
										snake_case       :     Dict =    self.get_tokenizer()
										snake_case       :     Any =    self.get_feature_extractor()
										snake_case       :     int =    self.get_decoder()

										snake_case       :     List[str] =    WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_,    feature_extractor=SCREAMING_SNAKE_CASE_,    decoder=SCREAMING_SNAKE_CASE_      )

										processor.save_pretrained(self.tmpdirname      )
										snake_case       :     Optional[int] =    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 	__snake_case							(							self	:				List[Any]      ):
										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
										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 	__snake_case							(							self	:				List[str]      ):
										snake_case       :     Union[str, Any] =    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 	__snake_case							(							self	:				List[Any]      ):
										snake_case       :     str =    self.get_feature_extractor()
										snake_case       :     Any =    self.get_tokenizer()
										snake_case       :     Tuple =    self.get_decoder()

										snake_case       :     int =    WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_,    feature_extractor=SCREAMING_SNAKE_CASE_,    decoder=SCREAMING_SNAKE_CASE_      )

										snake_case       :     Union[str, Any] =    floats_list((3, 10_00)      )

										snake_case       :     Tuple =    feature_extractor(SCREAMING_SNAKE_CASE_,    return_tensors='''np'''      )
										snake_case       :     List[str] =    processor(SCREAMING_SNAKE_CASE_,    return_tensors='''np'''      )

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




				def 	__snake_case							(							self	:				int      ):
										snake_case       :     str =    self.get_feature_extractor()
										snake_case       :     Tuple =    self.get_tokenizer()
										snake_case       :     List[str] =    self.get_decoder()

										snake_case       :     Union[str, Any] =    WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_,    feature_extractor=SCREAMING_SNAKE_CASE_,    decoder=SCREAMING_SNAKE_CASE_      )

										snake_case       :     Optional[Any] =    '''This is a test string'''

										snake_case       :     List[Any] =    processor(text=SCREAMING_SNAKE_CASE_      )

										snake_case       :     List[str] =    tokenizer(SCREAMING_SNAKE_CASE_      )

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




				def 	__snake_case							(							self	:				List[str],    SCREAMING_SNAKE_CASE_	:				Optional[int]=(2, 10, 16),    SCREAMING_SNAKE_CASE_	:				Optional[Any]=77      ):
										np.random.seed(SCREAMING_SNAKE_CASE_      )
										return np.random.rand(*SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				List[str]      ):
										snake_case       :     Tuple =    self.get_feature_extractor()
										snake_case       :     Any =    self.get_tokenizer()
										snake_case       :     int =    self.get_decoder()

										snake_case       :     List[Any] =    WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_,    feature_extractor=SCREAMING_SNAKE_CASE_,    decoder=SCREAMING_SNAKE_CASE_      )

										snake_case       :     List[str] =    self._get_dummy_logits(shape=(10, 16),    seed=13      )

										snake_case       :     Tuple =    processor.decode(SCREAMING_SNAKE_CASE_      )

										snake_case       :     Any =    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 	__snake_case							(							self	:				Tuple,    SCREAMING_SNAKE_CASE_	:				Optional[Any]      ):
										snake_case       :     Tuple =    self.get_feature_extractor()
										snake_case       :     Union[str, Any] =    self.get_tokenizer()
										snake_case       :     Any =    self.get_decoder()

										snake_case       :     str =    WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_,    feature_extractor=SCREAMING_SNAKE_CASE_,    decoder=SCREAMING_SNAKE_CASE_      )

										snake_case       :     Dict =    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:
																snake_case       :     Optional[int] =    processor.batch_decode(SCREAMING_SNAKE_CASE_      )
										else:
																with get_context(SCREAMING_SNAKE_CASE_      ).Pool() as pool:
																						snake_case       :     int =    processor.batch_decode(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_      )

										snake_case       :     Tuple =    list(SCREAMING_SNAKE_CASE_      )

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

										snake_case,					snake_case,					snake_case       :     List[str] =    [], [], []
										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 	__snake_case							(							self	:				int      ):
										snake_case       :     Any =    self.get_feature_extractor()
										snake_case       :     Union[str, Any] =    self.get_tokenizer()
										snake_case       :     str =    self.get_decoder()

										snake_case       :     Union[str, Any] =    WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_,    feature_extractor=SCREAMING_SNAKE_CASE_,    decoder=SCREAMING_SNAKE_CASE_      )

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

										snake_case       :     List[Any] =    15
										snake_case       :     int =    -20.0
										snake_case       :     str =    -4.0

										snake_case       :     Union[str, Any] =    processor.batch_decode(
										    SCREAMING_SNAKE_CASE_,    beam_width=SCREAMING_SNAKE_CASE_,    beam_prune_logp=SCREAMING_SNAKE_CASE_,    token_min_logp=SCREAMING_SNAKE_CASE_,    )
										snake_case       :     Any =    decoded_processor_out.text

										snake_case       :     List[str] =    list(SCREAMING_SNAKE_CASE_      )

										with get_context('''fork'''      ).Pool() as pool:
																snake_case       :     int =    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_,    )

										snake_case       :     List[str] =    [d[0][0] for d in decoded_decoder_out]
										snake_case       :     List[Any] =    [d[0][2] for d in decoded_decoder_out]
										snake_case       :     Optional[Any] =    [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.9474],    SCREAMING_SNAKE_CASE_,    atol=1e-3      )      )




				def 	__snake_case							(							self	:				Tuple      ):
										snake_case       :     Optional[Any] =    self.get_feature_extractor()
										snake_case       :     List[str] =    self.get_tokenizer()
										snake_case       :     Dict =    self.get_decoder()

										snake_case       :     str =    WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_,    feature_extractor=SCREAMING_SNAKE_CASE_,    decoder=SCREAMING_SNAKE_CASE_      )

										snake_case       :     List[Any] =    self._get_dummy_logits()

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

										snake_case       :     List[str] =    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_,    )
										snake_case       :     List[str] =    decoded_processor_out.text

										snake_case       :     str =    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:
																snake_case       :     int =    decoder.decode_beams_batch(
																    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    )

										snake_case       :     Tuple =    [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_      )
										snake_case       :     List[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,    SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Optional[Any]      ):
										snake_case       :     Union[str, Any] =    WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )

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

										snake_case       :     Dict =    os.listdir(SCREAMING_SNAKE_CASE_      )
										snake_case       :     Union[str, Any] =    ['''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 	__snake_case							(							self	:				int      ):
										snake_case       :     Dict =    snapshot_download('''hf-internal-testing/processor_with_lm'''      )

										snake_case       :     Union[str, Any] =    WavaVecaProcessorWithLM.from_pretrained(SCREAMING_SNAKE_CASE_      )

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

										snake_case       :     Union[str, Any] =    os.listdir(SCREAMING_SNAKE_CASE_      )
										snake_case       :     List[Any] =    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 	__snake_case							(							self	:				Optional[int]      ):
										snake_case       :     Any =    WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )
										snake_case       :     List[Any] =    AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''      )

										snake_case       :     Optional[Any] =    floats_list((3, 10_00)      )

										snake_case       :     Optional[int] =    processor_wavaveca(SCREAMING_SNAKE_CASE_,    return_tensors='''np'''      )
										snake_case       :     int =    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      )

										snake_case       :     List[Any] =    self._get_dummy_logits()

										snake_case       :     Dict =    processor_wavaveca.batch_decode(SCREAMING_SNAKE_CASE_      )
										snake_case       :     List[Any] =    processor_auto.batch_decode(SCREAMING_SNAKE_CASE_      )

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




				def 	__snake_case							(							self	:				List[Any]      ):
										snake_case       :     List[str] =    self.get_feature_extractor()
										snake_case       :     Dict =    self.get_tokenizer()
										snake_case       :     Optional[int] =    self.get_decoder()

										snake_case       :     Optional[Any] =    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 	__snake_case							(							SCREAMING_SNAKE_CASE_	:				List[str],    SCREAMING_SNAKE_CASE_	:				List[str]      ):
										snake_case       :     Dict =    [d[key] for d in offsets]
										return retrieved_list




				def 	__snake_case							(							self	:				Union[str, Any]      ):
										snake_case       :     List[str] =    WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )
										snake_case       :     str =    self._get_dummy_logits()[0]

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

										snake_case       :     Tuple =    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 	__snake_case							(							self	:				str      ):
										import torch

										snake_case       :     List[str] =    load_dataset('''common_voice''',    '''en''',    split='''train''',    streaming=SCREAMING_SNAKE_CASE_      )
										snake_case       :     Tuple =    ds.cast_column('''audio''',    datasets.Audio(sampling_rate=1_60_00      )      )
										snake_case       :     Union[str, Any] =    iter(SCREAMING_SNAKE_CASE_      )
										snake_case       :     Any =    next(SCREAMING_SNAKE_CASE_      )

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

										with torch.no_grad():
																snake_case       :     str =    model(SCREAMING_SNAKE_CASE_      ).logits.cpu().numpy()

										snake_case       :     Optional[int] =    processor.decode(logits[0],    output_word_offsets=SCREAMING_SNAKE_CASE_      )

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

										snake_case       :     int =    '''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
										snake_case       :     Optional[Any] =    torch.tensor(self.get_from_offsets(SCREAMING_SNAKE_CASE_,    '''start_time'''      )      )
										snake_case       :     Optional[Any] =    torch.tensor(self.get_from_offsets(SCREAMING_SNAKE_CASE_,    '''end_time'''      )      )

										# fmt: off
										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]      )
										snake_case       :     Any =    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(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    atol=0.01      )      )
										self.assertTrue(torch.allclose(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    atol=0.01      )      )



'''simple docstring'''





import pytest

from datasets import inspect_metric, list_metrics, load_metric

@pytest.fixture
def      A (       A_							:					str       ):
						monkeypatch.setattr('''datasets.utils.deprecation_utils._emitted_deprecation_warnings'''    ,							set()       )

@pytest.fixture
def      A (       A_							:					Optional[Any]       ):

						class 						a   :




										def __init__(							self	:				List[str],    SCREAMING_SNAKE_CASE_	:				Optional[Any]      ):
																snake_case       :     List[str] =    metric_id




						class 						a   :
										_snake_case		    = [MetricMock(__magic_name__    ) for metric_id in ['''accuracy''', '''mse''', '''precision''', '''codeparrot/apps_metric''']]




										def 	__snake_case							(							self	:				str      ):
																return self._metrics




						monkeypatch.setattr('''datasets.inspect.huggingface_hub'''    ,							HfhMock()       )





@pytest.mark.parametrize(
    '''func, args'''    ,							[(load_metric, ('''metrics/mse''',)), (list_metrics, ()), (inspect_metric, ('''metrics/mse''', '''tmp_path'''))]       )
def      A (       A_							:					str    ,							A_							:					Dict    ,							A_							:					int    ,							A_							:					int    ,							A_							:					str       ):
						if "tmp_path" in args:
												snake_case       :     List[str] =    tuple(arg if arg != '''tmp_path''' else tmp_path for arg in args       )
						with pytest.warns(A_    ,							match='''https://huggingface.co/docs/evaluate'''       ):
												func(*A_       )

