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






					def      _snake_case   (       self		):


											"""simple docstring"""





											lowercase_     :					str													=	'''| <pad> <unk> <s> </s> a b c d e f g h i j k'''.split()
											lowercase_     :					int													=	dict(zip(__SCREAMING_SNAKE_CASE    ,							range(len(__SCREAMING_SNAKE_CASE		)		)		)		)

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

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






					def      _snake_case   (       self    ,							**__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											lowercase_     :					str													=	self.add_kwargs_tokens_map.copy()
											kwargs.update(__SCREAMING_SNAKE_CASE		)
											return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname    ,							**__SCREAMING_SNAKE_CASE		)






					def      _snake_case   (       self    ,							**__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname    ,							**__SCREAMING_SNAKE_CASE		)






					def      _snake_case   (       self    ,							**__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name    ,							**__SCREAMING_SNAKE_CASE		)






					def      _snake_case   (       self		):


											"""simple docstring"""





											shutil.rmtree(self.tmpdirname		)






					def      _snake_case   (       self		):


											"""simple docstring"""





											lowercase_     :					Optional[Any]													=	self.get_tokenizer()
											lowercase_     :					List[str]													=	self.get_feature_extractor()
											lowercase_     :					Dict													=	self.get_decoder()

											lowercase_     :					Union[str, Any]													=	WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE    ,							feature_extractor=__SCREAMING_SNAKE_CASE    ,							decoder=__SCREAMING_SNAKE_CASE		)

											processor.save_pretrained(self.tmpdirname		)
											lowercase_     :					List[Any]													=	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		):


											"""simple docstring"""





											lowercase_     :					str													=	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
											lowercase_     :					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		):


											"""simple docstring"""





											lowercase_     :					Optional[int]													=	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		):


											"""simple docstring"""





											lowercase_     :					Tuple													=	self.get_feature_extractor()
											lowercase_     :					Optional[Any]													=	self.get_tokenizer()
											lowercase_     :					Tuple													=	self.get_decoder()

											lowercase_     :					Dict													=	WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE    ,							feature_extractor=__SCREAMING_SNAKE_CASE    ,							decoder=__SCREAMING_SNAKE_CASE		)

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

											lowercase_     :					Tuple													=	feature_extractor(__SCREAMING_SNAKE_CASE    ,							return_tensors='''np'''		)
											lowercase_     :					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		):


											"""simple docstring"""





											lowercase_     :					str													=	self.get_feature_extractor()
											lowercase_     :					Optional[int]													=	self.get_tokenizer()
											lowercase_     :					List[Any]													=	self.get_decoder()

											lowercase_     :					int													=	WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE    ,							feature_extractor=__SCREAMING_SNAKE_CASE    ,							decoder=__SCREAMING_SNAKE_CASE		)

											lowercase_     :					Tuple													=	'''This is a test string'''

											lowercase_     :					int													=	processor(text=__SCREAMING_SNAKE_CASE		)

											lowercase_     :					List[str]													=	tokenizer(__SCREAMING_SNAKE_CASE		)

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






					def      _snake_case   (       self    ,							__SCREAMING_SNAKE_CASE=(2, 10, 16)    ,							__SCREAMING_SNAKE_CASE=77		):


											"""simple docstring"""





											np.random.seed(__SCREAMING_SNAKE_CASE		)
											return np.random.rand(*__SCREAMING_SNAKE_CASE		)






					def      _snake_case   (       self		):


											"""simple docstring"""





											lowercase_     :					Union[str, Any]													=	self.get_feature_extractor()
											lowercase_     :					Union[str, Any]													=	self.get_tokenizer()
											lowercase_     :					List[Any]													=	self.get_decoder()

											lowercase_     :					Tuple													=	WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE    ,							feature_extractor=__SCREAMING_SNAKE_CASE    ,							decoder=__SCREAMING_SNAKE_CASE		)

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

											lowercase_     :					List[Any]													=	processor.decode(__SCREAMING_SNAKE_CASE		)

											lowercase_     :					List[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    ,							__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											lowercase_     :					Optional[int]													=	self.get_feature_extractor()
											lowercase_     :					Dict													=	self.get_tokenizer()
											lowercase_     :					List[Any]													=	self.get_decoder()

											lowercase_     :					Optional[Any]													=	WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE    ,							feature_extractor=__SCREAMING_SNAKE_CASE    ,							decoder=__SCREAMING_SNAKE_CASE		)

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

											lowercase_     :					str													=	list(__SCREAMING_SNAKE_CASE		)

											with get_context('''fork'''		).Pool() as p:
																	lowercase_     :					List[Any]													=	decoder.decode_beams_batch(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE		)

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

											self.assertListEqual(__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		):


											"""simple docstring"""





											lowercase_     :					Union[str, Any]													=	self.get_feature_extractor()
											lowercase_     :					int													=	self.get_tokenizer()
											lowercase_     :					Union[str, Any]													=	self.get_decoder()

											lowercase_     :					int													=	WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE    ,							feature_extractor=__SCREAMING_SNAKE_CASE    ,							decoder=__SCREAMING_SNAKE_CASE		)

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

											lowercase_     :					Any													=	15
											lowercase_     :					Any													=	-20.0
											lowercase_     :					Tuple													=	-4.0

											lowercase_     :					str													=	processor.batch_decode(
											    __SCREAMING_SNAKE_CASE    ,							beam_width=__SCREAMING_SNAKE_CASE    ,							beam_prune_logp=__SCREAMING_SNAKE_CASE    ,							token_min_logp=__SCREAMING_SNAKE_CASE    ,							)
											lowercase_     :					Dict													=	decoded_processor_out.text

											lowercase_     :					List[Any]													=	list(__SCREAMING_SNAKE_CASE		)

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

											lowercase_     :					int													=	[d[0][0] for d in decoded_decoder_out]
											lowercase_     :					str													=	[d[0][2] for d in decoded_decoder_out]
											lowercase_     :					List[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.9_474]    ,							__SCREAMING_SNAKE_CASE    ,							atol=1E-3		)		)






					def      _snake_case   (       self		):


											"""simple docstring"""





											lowercase_     :					List[str]													=	self.get_feature_extractor()
											lowercase_     :					List[str]													=	self.get_tokenizer()
											lowercase_     :					int													=	self.get_decoder()

											lowercase_     :					List[Any]													=	WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE    ,							feature_extractor=__SCREAMING_SNAKE_CASE    ,							decoder=__SCREAMING_SNAKE_CASE		)

											lowercase_     :					List[str]													=	self._get_dummy_logits()

											lowercase_     :					Tuple													=	2.0
											lowercase_     :					Optional[int]													=	5.0
											lowercase_     :					int													=	-20.0
											lowercase_     :					str													=	True

											lowercase_     :					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    ,							)
											lowercase_     :					str													=	decoded_processor_out.text

											lowercase_     :					Dict													=	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:
																	lowercase_     :					int													=	decoder.decode_beams_batch(
																	    __SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE    ,							)

											lowercase_     :					str													=	[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		)
											lowercase_     :					List[str]													=	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		):


											"""simple docstring"""





											lowercase_     :					int													=	WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''		)

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

											lowercase_     :					str													=	os.listdir(__SCREAMING_SNAKE_CASE		)
											lowercase_     :					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		):


											"""simple docstring"""





											lowercase_     :					List[Any]													=	snapshot_download('''hf-internal-testing/processor_with_lm'''		)

											lowercase_     :					Union[str, Any]													=	WavaVecaProcessorWithLM.from_pretrained(__SCREAMING_SNAKE_CASE		)

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

											lowercase_     :					str													=	os.listdir(__SCREAMING_SNAKE_CASE		)
											lowercase_     :					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		):


											"""simple docstring"""





											lowercase_     :					Optional[int]													=	WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''		)
											lowercase_     :					Dict													=	AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''		)

											lowercase_     :					List[Any]													=	floats_list((3, 10_00)		)

											lowercase_     :					Union[str, Any]													=	processor_wavaveca(__SCREAMING_SNAKE_CASE    ,							return_tensors='''np'''		)
											lowercase_     :					List[Any]													=	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		)

											lowercase_     :					Any													=	self._get_dummy_logits()

											lowercase_     :					Tuple													=	processor_wavaveca.batch_decode(__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Tuple													=	processor_auto.batch_decode(__SCREAMING_SNAKE_CASE		)

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






					def      _snake_case   (       self		):


											"""simple docstring"""





											lowercase_     :					Optional[int]													=	self.get_feature_extractor()
											lowercase_     :					Optional[int]													=	self.get_tokenizer()
											lowercase_     :					int													=	self.get_decoder()

											lowercase_     :					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    ,							__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											lowercase_     :					Dict													=	[d[key] for d in offsets]
											return retrieved_list






					def      _snake_case   (       self		):


											"""simple docstring"""





											lowercase_     :					str													=	WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''		)
											lowercase_     :					List[str]													=	self._get_dummy_logits()[0]

											lowercase_     :					Tuple													=	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		):


											"""simple docstring"""





											lowercase_     :					Dict													=	WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''		)
											lowercase_     :					List[Any]													=	self._get_dummy_logits()

											lowercase_     :					Optional[Any]													=	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		):


											"""simple docstring"""





											import torch

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

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

											with torch.no_grad():
																	lowercase_     :					Any													=	model(__SCREAMING_SNAKE_CASE		).logits.cpu().numpy()

											lowercase_     :					Tuple													=	processor.decode(logits[0]    ,							output_word_offsets=__SCREAMING_SNAKE_CASE		)

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

											lowercase_     :					Union[str, Any]													=	'''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
											lowercase_     :					Optional[Any]													=	torch.tensor(self.get_from_offsets(__SCREAMING_SNAKE_CASE    ,							'''start_time'''		)		)
											lowercase_     :					Union[str, Any]													=	torch.tensor(self.get_from_offsets(__SCREAMING_SNAKE_CASE    ,							'''end_time'''		)		)

											# fmt: off
											lowercase_     :					Any													=	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]		)
											lowercase_     :					List[str]													=	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		)		)





'''simple docstring'''







import warnings

from ...utils import logging
from .image_processing_donut import DonutImageProcessor


_lowercase			:  Any   							=	logging.get_logger(__name__)






class 			lowerCAmelCase__       (  lowerCamelCase_					):






					def __init__(       self    ,							*__SCREAMING_SNAKE_CASE    ,							**__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											warnings.warn(
											    '''The class DonutFeatureExtractor is deprecated and will be removed in version 5 of Transformers. Please'''
											    ''' use DonutImageProcessor instead.'''    ,							__SCREAMING_SNAKE_CASE    ,							)
											super().__init__(*__SCREAMING_SNAKE_CASE    ,							**__SCREAMING_SNAKE_CASE		)



