





import itertools
import random
import unittest

import numpy as np

from transformers import BatchFeature, SpeechTaFeatureExtractor
from transformers.testing_utils import require_torch
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
	import torch


_SCREAMING_SNAKE_CASE					=    random.Random()




def    _lowerCAmelCase						(				lowerCamelCase_			:  str		,       lowerCamelCase_			:  Any=1.0		,       lowerCamelCase_			:  List[str]=None		,       lowerCamelCase_			:  Optional[int]=None				):
			if rng is None:
						__lowercase										=							global_rng

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

			return values






@require_torch
class 							__lowercase       (      unittest.TestCase						):







	'''simple docstring'''

	def __init__(self     ,_lowerCamelCase     ,_lowerCamelCase=7     ,_lowerCamelCase=400     ,_lowerCamelCase=2000     ,_lowerCamelCase=1     ,_lowerCamelCase=0.0     ,_lowerCamelCase=16000     ,_lowerCamelCase=True     ,_lowerCamelCase=80     ,_lowerCamelCase=16     ,_lowerCamelCase=64     ,_lowerCamelCase="hann_window"     ,_lowerCamelCase=80     ,_lowerCamelCase=7600     ,_lowerCamelCase=1E-1_0     ,_lowerCamelCase=True     ,)							->							Any:







				'''simple docstring'''





				__lowercase										=							parent
				__lowercase										=							batch_size
				__lowercase										=							min_seq_length
				__lowercase										=							max_seq_length
				__lowercase										=							(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
				__lowercase										=							feature_size
				__lowercase										=							padding_value
				__lowercase										=							sampling_rate
				__lowercase										=							do_normalize
				__lowercase										=							num_mel_bins
				__lowercase										=							hop_length
				__lowercase										=							win_length
				__lowercase										=							win_function
				__lowercase										=							fmin
				__lowercase										=							fmax
				__lowercase										=							mel_floor
				__lowercase										=							return_attention_mask


	def 						_UpperCAmelCase     (self )							->							List[str]:







				'''simple docstring'''





				return {
				    "feature_size": self.feature_size,
				    "padding_value": self.padding_value,
				    "sampling_rate": self.sampling_rate,
				    "do_normalize": self.do_normalize,
				    "num_mel_bins": self.num_mel_bins,
				    "hop_length": self.hop_length,
				    "win_length": self.win_length,
				    "win_function": self.win_function,
				    "fmin": self.fmin,
				    "fmax": self.fmax,
				    "mel_floor": self.mel_floor,
				    "return_attention_mask": self.return_attention_mask,
				}


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase=False     ,_lowerCamelCase=False )							->							Union[str, Any]:







				'''simple docstring'''





				def _flatten(_lowerCamelCase ):
							return list(itertools.chain(*_snake_case ) )

				if equal_length:
							__lowercase										=							floats_list((self.batch_size, self.max_seq_length) )
				else:
							# make sure that inputs increase in size
							__lowercase										=							[
							    _flatten(floats_list((x, self.feature_size) ) )
							    for x in range(self.min_seq_length     ,self.max_seq_length     ,self.seq_length_diff )
							]

				if numpify:
							__lowercase										=							[np.asarray(_snake_case ) for x in speech_inputs]

				return speech_inputs


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase=False     ,_lowerCamelCase=False )							->							List[Any]:







				'''simple docstring'''





				if equal_length:
							__lowercase										=							[floats_list((self.max_seq_length, self.num_mel_bins) ) for _ in range(self.batch_size )]
				else:
							# make sure that inputs increase in size
							__lowercase										=							[
							    floats_list((x, self.num_mel_bins) )
							    for x in range(self.min_seq_length     ,self.max_seq_length     ,self.seq_length_diff )
							]

				if numpify:
							__lowercase										=							[np.asarray(_snake_case ) for x in speech_inputs]

				return speech_inputs






@require_torch
class 							__lowercase       (      UpperCAmelCase_ ,      unittest.TestCase						):







	'''simple docstring'''

	a    :				Dict											=       SpeechTaFeatureExtractor


	def 						_UpperCAmelCase     (self )							->							int:







				'''simple docstring'''





				__lowercase										=							SpeechTaFeatureExtractionTester(self )


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							str:







				'''simple docstring'''





				self.assertTrue(np.all(np.mean(_snake_case     ,axis=0 ) < 1E-3 ) )
				self.assertTrue(np.all(np.abs(np.var(_snake_case     ,axis=0 ) - 1 ) < 1E-3 ) )


	def 						_UpperCAmelCase     (self )							->							int:







				'''simple docstring'''





				__lowercase										=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
				# create three inputs of length 800, 1000, and 1200
				__lowercase										=							[floats_list((1, x) )[0] for x in range(800     ,1400     ,200 )]
				__lowercase										=							[np.asarray(_snake_case ) for speech_input in speech_inputs]

				# Test not batched input
				__lowercase										=							feat_extract(speech_inputs[0]     ,return_tensors='''np''' ).input_values
				__lowercase										=							feat_extract(np_speech_inputs[0]     ,return_tensors='''np''' ).input_values
				self.assertTrue(np.allclose(_snake_case     ,_snake_case     ,atol=1E-3 ) )

				# Test batched
				__lowercase										=							feat_extract(_snake_case     ,return_tensors='''np''' ).input_values
				__lowercase										=							feat_extract(_snake_case     ,return_tensors='''np''' ).input_values
				for enc_seq_a, enc_seq_a in zip(_snake_case     ,_snake_case ):
							self.assertTrue(np.allclose(_snake_case     ,_snake_case     ,atol=1E-3 ) )


	def 						_UpperCAmelCase     (self )							->							List[Any]:







				'''simple docstring'''





				__lowercase										=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
				__lowercase										=							[floats_list((1, x) )[0] for x in range(800     ,1400     ,200 )]

				__lowercase										=							['''longest''', '''max_length''', '''do_not_pad''']
				__lowercase										=							[None, 1600, None]
				for max_length, padding in zip(_snake_case     ,_snake_case ):
							__lowercase										=							feat_extract(_snake_case     ,padding=_snake_case     ,max_length=_snake_case     ,return_tensors='''np''' )
							__lowercase										=							processed.input_values

							self._check_zero_mean_unit_variance(input_values[0][:800] )
							self.assertTrue(input_values[0][800:].sum() < 1E-6 )
							self._check_zero_mean_unit_variance(input_values[1][:1000] )
							self.assertTrue(input_values[0][1000:].sum() < 1E-6 )
							self._check_zero_mean_unit_variance(input_values[2][:1200] )


	def 						_UpperCAmelCase     (self )							->							Dict:







				'''simple docstring'''





				__lowercase										=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
				__lowercase										=							range(800     ,1400     ,200 )
				__lowercase										=							[floats_list((1, x) )[0] for x in lengths]

				__lowercase										=							['''longest''', '''max_length''', '''do_not_pad''']
				__lowercase										=							[None, 1600, None]

				for max_length, padding in zip(_snake_case     ,_snake_case ):
							__lowercase										=							feat_extract(_snake_case     ,max_length=_snake_case     ,padding=_snake_case )
							__lowercase										=							processed.input_values

							self._check_zero_mean_unit_variance(input_values[0][:800] )
							self._check_zero_mean_unit_variance(input_values[1][:1000] )
							self._check_zero_mean_unit_variance(input_values[2][:1200] )


	def 						_UpperCAmelCase     (self )							->							List[str]:







				'''simple docstring'''





				__lowercase										=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
				__lowercase										=							[floats_list((1, x) )[0] for x in range(800     ,1400     ,200 )]
				__lowercase										=							feat_extract(
				    _snake_case     ,truncation=_snake_case     ,max_length=1000     ,padding='''max_length'''     ,return_tensors='''np''' )
				__lowercase										=							processed.input_values

				self._check_zero_mean_unit_variance(input_values[0, :800] )
				self._check_zero_mean_unit_variance(input_values[1] )
				self._check_zero_mean_unit_variance(input_values[2] )


	def 						_UpperCAmelCase     (self )							->							List[Any]:







				'''simple docstring'''





				__lowercase										=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
				__lowercase										=							[floats_list((1, x) )[0] for x in range(800     ,1400     ,200 )]
				__lowercase										=							feat_extract(
				    _snake_case     ,truncation=_snake_case     ,max_length=1000     ,padding='''longest'''     ,return_tensors='''np''' )
				__lowercase										=							processed.input_values

				self._check_zero_mean_unit_variance(input_values[0, :800] )
				self._check_zero_mean_unit_variance(input_values[1, :1000] )
				self._check_zero_mean_unit_variance(input_values[2] )

				# make sure that if max_length < longest -> then pad to max_length
				self.assertTrue(input_values.shape == (3, 1000) )

				__lowercase										=							[floats_list((1, x) )[0] for x in range(800     ,1400     ,200 )]
				__lowercase										=							feat_extract(
				    _snake_case     ,truncation=_snake_case     ,max_length=2000     ,padding='''longest'''     ,return_tensors='''np''' )
				__lowercase										=							processed.input_values

				self._check_zero_mean_unit_variance(input_values[0, :800] )
				self._check_zero_mean_unit_variance(input_values[1, :1000] )
				self._check_zero_mean_unit_variance(input_values[2] )

				# make sure that if max_length > longest -> then pad to longest
				self.assertTrue(input_values.shape == (3, 1200) )


	def 						_UpperCAmelCase     (self )							->							List[str]:







				'''simple docstring'''





				__lowercase										=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
				__lowercase										=							np.random.rand(100 ).astype(np.floataa )
				__lowercase										=							np_speech_inputs.tolist()

				for inputs in [py_speech_inputs, np_speech_inputs]:
							__lowercase										=							feature_extractor.pad([{'''input_values''': inputs}]     ,return_tensors='''np''' )
							self.assertTrue(np_processed.input_values.dtype == np.floataa )
							__lowercase										=							feature_extractor.pad([{'''input_values''': inputs}]     ,return_tensors='''pt''' )
							self.assertTrue(pt_processed.input_values.dtype == torch.floataa )


	def 						_UpperCAmelCase     (self )							->							Optional[Any]:







				'''simple docstring'''





				__lowercase										=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
				# create three inputs of length 800, 1000, and 1200
				__lowercase										=							[floats_list((1, x) )[0] for x in range(800     ,1400     ,200 )]
				__lowercase										=							[np.asarray(_snake_case ) for speech_input in speech_inputs]

				# Test feature size
				__lowercase										=							feature_extractor(audio_target=_snake_case     ,padding=_snake_case     ,return_tensors='''np''' ).input_values
				self.assertTrue(input_values.ndim == 3 )
				self.assertTrue(input_values.shape[-1] == feature_extractor.num_mel_bins )

				# Test not batched input
				__lowercase										=							feature_extractor(speech_inputs[0]     ,return_tensors='''np''' ).input_values
				__lowercase										=							feature_extractor(np_speech_inputs[0]     ,return_tensors='''np''' ).input_values
				self.assertTrue(np.allclose(_snake_case     ,_snake_case     ,atol=1E-3 ) )

				# Test batched
				__lowercase										=							feature_extractor(_snake_case     ,return_tensors='''np''' ).input_values
				__lowercase										=							feature_extractor(_snake_case     ,return_tensors='''np''' ).input_values
				for enc_seq_a, enc_seq_a in zip(_snake_case     ,_snake_case ):
							self.assertTrue(np.allclose(_snake_case     ,_snake_case     ,atol=1E-3 ) )

				# Test 2-D numpy arrays are batched.
				__lowercase										=							[floats_list((1, x) )[0] for x in (800, 800, 800)]
				__lowercase										=							np.asarray(_snake_case )
				__lowercase										=							feature_extractor(_snake_case     ,return_tensors='''np''' ).input_values
				__lowercase										=							feature_extractor(_snake_case     ,return_tensors='''np''' ).input_values
				for enc_seq_a, enc_seq_a in zip(_snake_case     ,_snake_case ):
							self.assertTrue(np.allclose(_snake_case     ,_snake_case     ,atol=1E-3 ) )


	def 						_UpperCAmelCase     (self )							->							Dict:







				'''simple docstring'''





				__lowercase										=							self.feat_extract_tester.prepare_inputs_for_target()
				__lowercase										=							self.feature_extraction_class(**self.feat_extract_dict )
				__lowercase										=							feat_extract.model_input_names[0]

				__lowercase										=							BatchFeature({input_name: speech_inputs} )

				self.assertTrue(all(len(_snake_case ) == len(_snake_case ) for x, y in zip(_snake_case     ,processed_features[input_name] ) ) )

				__lowercase										=							self.feat_extract_tester.prepare_inputs_for_target(equal_length=_snake_case )
				__lowercase										=							BatchFeature({input_name: speech_inputs}     ,tensor_type='''np''' )

				__lowercase										=							processed_features[input_name]

				if len(batch_features_input.shape ) < 3:
							__lowercase										=							batch_features_input[:, :, None]

				self.assertTrue(
				    batch_features_input.shape
				    == (self.feat_extract_tester.batch_size, len(speech_inputs[0] ), self.feat_extract_tester.num_mel_bins) )


	@require_torch
	def 						_UpperCAmelCase     (self )							->							Tuple:







				'''simple docstring'''





				__lowercase										=							self.feat_extract_tester.prepare_inputs_for_target(equal_length=_snake_case )
				__lowercase										=							self.feature_extraction_class(**self.feat_extract_dict )
				__lowercase										=							feat_extract.model_input_names[0]

				__lowercase										=							BatchFeature({input_name: speech_inputs}     ,tensor_type='''pt''' )

				__lowercase										=							processed_features[input_name]

				if len(batch_features_input.shape ) < 3:
							__lowercase										=							batch_features_input[:, :, None]

				self.assertTrue(
				    batch_features_input.shape
				    == (self.feat_extract_tester.batch_size, len(speech_inputs[0] ), self.feat_extract_tester.num_mel_bins) )


	@require_torch
	def 						_UpperCAmelCase     (self )							->							List[Any]:







				'''simple docstring'''





				__lowercase										=							self.feature_extraction_class(**self.feat_extract_dict )
				__lowercase										=							self.feat_extract_tester.prepare_inputs_for_target()
				__lowercase										=							feat_extract.model_input_names[0]

				__lowercase										=							BatchFeature({input_name: speech_inputs} )

				__lowercase										=							feat_extract.num_mel_bins  # hack!

				__lowercase										=							feat_extract.pad(_snake_case     ,padding='''longest'''     ,return_tensors='''np''' )[input_name]
				__lowercase										=							feat_extract.pad(_snake_case     ,padding='''longest'''     ,return_tensors='''pt''' )[input_name]

				self.assertTrue(abs(input_np.astype(np.floataa ).sum() - input_pt.numpy().astype(np.floataa ).sum() ) < 1E-2 )


	def 						_UpperCAmelCase     (self )							->							Dict:







				'''simple docstring'''





				__lowercase										=							self.feat_extract_dict
				__lowercase										=							True
				__lowercase										=							self.feature_extraction_class(**_snake_case )
				__lowercase										=							self.feat_extract_tester.prepare_inputs_for_target()
				__lowercase										=							[len(_snake_case ) for x in speech_inputs]
				__lowercase										=							feat_extract.model_input_names[0]

				__lowercase										=							BatchFeature({input_name: speech_inputs} )

				__lowercase										=							feat_extract.num_mel_bins  # hack!

				__lowercase										=							feat_extract.pad(_snake_case     ,padding='''longest'''     ,return_tensors='''np''' )
				self.assertIn('''attention_mask'''     ,_snake_case )
				self.assertListEqual(list(processed.attention_mask.shape )     ,list(processed[input_name].shape[:2] ) )
				self.assertListEqual(processed.attention_mask.sum(-1 ).tolist()     ,_snake_case )


	def 						_UpperCAmelCase     (self )							->							Any:







				'''simple docstring'''





				__lowercase										=							self.feat_extract_dict
				__lowercase										=							True
				__lowercase										=							self.feature_extraction_class(**_snake_case )
				__lowercase										=							self.feat_extract_tester.prepare_inputs_for_target()
				__lowercase										=							[len(_snake_case ) for x in speech_inputs]
				__lowercase										=							feat_extract.model_input_names[0]

				__lowercase										=							BatchFeature({input_name: speech_inputs} )
				__lowercase										=							min(_snake_case )

				__lowercase										=							feat_extract.num_mel_bins  # hack!

				__lowercase										=							feat_extract.pad(
				    _snake_case     ,padding='''max_length'''     ,max_length=_snake_case     ,truncation=_snake_case     ,return_tensors='''np''' )
				self.assertIn('''attention_mask'''     ,_snake_case )
				self.assertListEqual(
				    list(processed_pad.attention_mask.shape )     ,[processed_pad[input_name].shape[0], max_length] )
				self.assertListEqual(
				    processed_pad.attention_mask[:, :max_length].sum(-1 ).tolist()     ,[max_length for x in speech_inputs] )


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							Tuple:







				'''simple docstring'''





				from datasets import load_dataset

				__lowercase										=							load_dataset('''hf-internal-testing/librispeech_asr_dummy'''     ,'''clean'''     ,split='''validation''' )
				# automatic decoding with librispeech
				__lowercase										=							ds.sort('''id''' ).select(range(_snake_case ) )[:num_samples]['''audio''']

				return [x["array"] for x in speech_samples]


	def 						_UpperCAmelCase     (self )							->							Optional[int]:







				'''simple docstring'''





				__lowercase										=							torch.tensor(
				    [2.3_8_0_4E-0_3, 2.0_7_5_2E-0_3, 1.9_8_3_6E-0_3, 2.1_0_5_7E-0_3, 1.6_1_7_4E-0_3,
				     3.0_5_1_8E-0_4, 9.1_5_5_3E-0_5, 3.3_5_6_9E-0_4, 9.7_6_5_6E-0_4, 1.8_3_1_1E-0_3,
				     2.0_1_4_2E-0_3, 2.1_0_5_7E-0_3, 1.7_3_9_5E-0_3, 4.5_7_7_6E-0_4, -3.9_6_7_3E-0_4,
				     4.5_7_7_6E-0_4, 1.0_0_7_1E-0_3, 9.1_5_5_3E-0_5, 4.8_8_2_8E-0_4, 1.1_5_9_7E-0_3,
				     7.3_2_4_2E-0_4, 9.4_6_0_4E-0_4, 1.8_0_0_5E-0_3, 1.8_3_1_1E-0_3, 8.8_5_0_1E-0_4,
				     4.2_7_2_5E-0_4, 4.8_8_2_8E-0_4, 7.3_2_4_2E-0_4, 1.0_9_8_6E-0_3, 2.1_0_5_7E-0_3] )
				# fmt: on

				__lowercase										=							self._load_datasamples(1 )
				__lowercase										=							SpeechTaFeatureExtractor()
				__lowercase										=							feature_extractor(_snake_case     ,return_tensors='''pt''' ).input_values
				self.assertEquals(input_values.shape     ,(1, 93680) )
				self.assertTrue(torch.allclose(input_values[0, :30]     ,_snake_case     ,atol=1E-6 ) )


	def 						_UpperCAmelCase     (self )							->							List[str]:







				'''simple docstring'''





				__lowercase										=							torch.tensor(
				    [-2.6_8_7_0, -3.0_1_0_4, -3.1_3_5_6, -3.5_3_5_2, -3.0_0_4_4, -3.0_3_5_3, -3.4_7_1_9, -3.6_7_7_7,
				     -3.1_5_2_0, -2.9_4_3_5, -2.6_5_5_3, -2.8_7_9_5, -2.9_9_4_4, -2.5_9_2_1, -3.0_2_7_9, -3.0_3_8_6,
				     -3.0_8_6_4, -3.1_2_9_1, -3.2_3_5_3, -2.7_4_4_4, -2.6_8_3_1, -2.7_2_8_7, -3.1_7_6_1, -3.1_5_7_1,
				     -3.2_7_2_6, -3.0_5_8_2, -3.1_0_0_7, -3.4_5_3_3, -3.4_6_9_5, -3.0_9_9_8] )
				# fmt: on

				__lowercase										=							self._load_datasamples(1 )
				__lowercase										=							SpeechTaFeatureExtractor()
				__lowercase										=							feature_extractor(audio_target=_snake_case     ,return_tensors='''pt''' ).input_values
				self.assertEquals(input_values.shape     ,(1, 366, 80) )
				self.assertTrue(torch.allclose(input_values[0, 0, :30]     ,_snake_case     ,atol=1E-4 ) )












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







	'''simple docstring'''

	def 						_UpperCAmelCase     (self )							->							Union[str, Any]:







				'''simple docstring'''





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

				__lowercase										=							{
				    '''unk_token''': '''<unk>''',
				    '''bos_token''': '''<s>''',
				    '''eos_token''': '''</s>''',
				}
				__lowercase										=							{
				    '''feature_size''': 1,
				    '''padding_value''': 0.0,
				    '''sampling_rate''': 16000,
				    '''return_attention_mask''': False,
				    '''do_normalize''': True,
				}

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

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

				# load decoder from hub
				__lowercase										=							'''hf-internal-testing/ngram-beam-search-decoder'''


	def 						_UpperCAmelCase     (self     ,**_lowerCamelCase )							->							List[str]:







				'''simple docstring'''





				__lowercase										=							self.add_kwargs_tokens_map.copy()
				kwargs.update(_lowerCamelCase )
				return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname     ,**_lowerCamelCase )


	def 						_UpperCAmelCase     (self     ,**_lowerCamelCase )							->							List[Any]:







				'''simple docstring'''





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


	def 						_UpperCAmelCase     (self     ,**_lowerCamelCase )							->							Dict:







				'''simple docstring'''





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


	def 						_UpperCAmelCase     (self )							->							Optional[Any]:







				'''simple docstring'''





				shutil.rmtree(self.tmpdirname )


	def 						_UpperCAmelCase     (self )							->							Optional[Any]:







				'''simple docstring'''





				__lowercase										=							self.get_tokenizer()
				__lowercase										=							self.get_feature_extractor()
				__lowercase										=							self.get_decoder()

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

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

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

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

				# decoder
				self.assertEqual(processor.decoder._alphabet.labels     ,decoder._alphabet.labels )
				self.assertEqual(
				    processor.decoder.model_container[decoder._model_key]._unigram_set     ,decoder.model_container[decoder._model_key]._unigram_set     ,)
				self.assertIsInstance(processor.decoder     ,_lowerCamelCase )


	def 						_UpperCAmelCase     (self )							->							Any:







				'''simple docstring'''





				__lowercase										=							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										=							WavaVecaProcessorWithLM.from_pretrained(
				    self.tmpdirname     ,alpha=5.0     ,beta=3.0     ,score_boundary=-7.0     ,unk_score_offset=3 )

				# decoder
				self.assertEqual(processor.language_model.alpha     ,5.0 )
				self.assertEqual(processor.language_model.beta     ,3.0 )
				self.assertEqual(processor.language_model.score_boundary     ,-7.0 )
				self.assertEqual(processor.language_model.unk_score_offset     ,3 )


	def 						_UpperCAmelCase     (self )							->							Dict:







				'''simple docstring'''





				__lowercase										=							self.get_tokenizer()
				# add token to trigger raise
				tokenizer.add_tokens(['''xx'''] )
				with self.assertRaisesRegex(_lowerCamelCase     ,'''include''' ):
							WavaVecaProcessorWithLM(
							    tokenizer=_lowerCamelCase     ,feature_extractor=self.get_feature_extractor()     ,decoder=self.get_decoder() )


	def 						_UpperCAmelCase     (self )							->							Optional[int]:







				'''simple docstring'''





				__lowercase										=							self.get_feature_extractor()
				__lowercase										=							self.get_tokenizer()
				__lowercase										=							self.get_decoder()

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

				__lowercase										=							floats_list((3, 1000) )

				__lowercase										=							feature_extractor(_lowerCamelCase     ,return_tensors='''np''' )
				__lowercase										=							processor(_lowerCamelCase     ,return_tensors='''np''' )

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


	def 						_UpperCAmelCase     (self )							->							List[Any]:







				'''simple docstring'''





				__lowercase										=							self.get_feature_extractor()
				__lowercase										=							self.get_tokenizer()
				__lowercase										=							self.get_decoder()

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

				__lowercase										=							'''This is a test string'''

				__lowercase										=							processor(text=_lowerCamelCase )

				__lowercase										=							tokenizer(_lowerCamelCase )

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


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase=(2, 10, 16)     ,_lowerCamelCase=77 )							->							Optional[int]:







				'''simple docstring'''





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


	def 						_UpperCAmelCase     (self )							->							str:







				'''simple docstring'''





				__lowercase										=							self.get_feature_extractor()
				__lowercase										=							self.get_tokenizer()
				__lowercase										=							self.get_decoder()

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

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

				__lowercase										=							processor.decode(_lowerCamelCase )

				__lowercase										=							decoder.decode_beams(_lowerCamelCase )[0]

				self.assertEqual(decoded_decoder[0]     ,decoded_processor.text )
				self.assertEqual('''</s> <s> </s>'''     ,decoded_processor.text )
				self.assertEqual(decoded_decoder[-2]     ,decoded_processor.logit_score )
				self.assertEqual(decoded_decoder[-1]     ,decoded_processor.lm_score )


	@parameterized.expand([[None], ['''fork'''], ['''spawn''']] )
	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							Optional[Any]:







				'''simple docstring'''





				__lowercase										=							self.get_feature_extractor()
				__lowercase										=							self.get_tokenizer()
				__lowercase										=							self.get_decoder()

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

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

				__lowercase										=							list(_lowerCamelCase )

				with get_context('''fork''' ).Pool() as p:
							__lowercase										=							decoder.decode_beams_batch(_lowerCamelCase     ,_lowerCamelCase )

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

				self.assertListEqual(_lowerCamelCase     ,decoded_processor.text )
				self.assertListEqual(['''<s> <s> </s>''', '''<s> <s> <s>''']     ,decoded_processor.text )
				self.assertListEqual(_lowerCamelCase     ,decoded_processor.logit_score )
				self.assertListEqual(_lowerCamelCase     ,decoded_processor.lm_score )


	def 						_UpperCAmelCase     (self )							->							Any:







				'''simple docstring'''





				__lowercase										=							self.get_feature_extractor()
				__lowercase										=							self.get_tokenizer()
				__lowercase										=							self.get_decoder()

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

				__lowercase										=							self._get_dummy_logits()

				__lowercase										=							15
				__lowercase										=							-2_0.0
				__lowercase										=							-4.0

				__lowercase										=							processor.batch_decode(
				    _lowerCamelCase     ,beam_width=_lowerCamelCase     ,beam_prune_logp=_lowerCamelCase     ,token_min_logp=_lowerCamelCase     ,)
				__lowercase										=							decoded_processor_out.text

				__lowercase										=							list(_lowerCamelCase )

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

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

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

				self.assertTrue(np.array_equal(_lowerCamelCase     ,decoded_processor_out.logit_score ) )
				self.assertTrue(np.allclose([-2_0.0_5_4, -1_8.4_4_7]     ,_lowerCamelCase     ,atol=1E-3 ) )

				self.assertTrue(np.array_equal(_lowerCamelCase     ,decoded_processor_out.lm_score ) )
				self.assertTrue(np.allclose([-1_5.5_5_4, -1_3.9_4_7_4]     ,_lowerCamelCase     ,atol=1E-3 ) )


	def 						_UpperCAmelCase     (self )							->							Union[str, Any]:







				'''simple docstring'''





				__lowercase										=							self.get_feature_extractor()
				__lowercase										=							self.get_tokenizer()
				__lowercase										=							self.get_decoder()

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

				__lowercase										=							self._get_dummy_logits()

				__lowercase										=							2.0
				__lowercase										=							5.0
				__lowercase										=							-2_0.0
				__lowercase										=							True

				__lowercase										=							processor.batch_decode(
				    _lowerCamelCase     ,alpha=_lowerCamelCase     ,beta=_lowerCamelCase     ,unk_score_offset=_lowerCamelCase     ,lm_score_boundary=_lowerCamelCase     ,)
				__lowercase										=							decoded_processor_out.text

				__lowercase										=							list(_lowerCamelCase )
				decoder.reset_params(
				    alpha=_lowerCamelCase     ,beta=_lowerCamelCase     ,unk_score_offset=_lowerCamelCase     ,lm_score_boundary=_lowerCamelCase     ,)

				with get_context('''fork''' ).Pool() as pool:
							__lowercase										=							decoder.decode_beams_batch(
							    _lowerCamelCase     ,_lowerCamelCase     ,)

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

				self.assertListEqual(_lowerCamelCase     ,_lowerCamelCase )
				self.assertListEqual(['''<s> </s> <s> </s> </s>''', '''</s> </s> <s> </s> </s>''']     ,_lowerCamelCase )
				__lowercase										=							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     ,-2_0.0 )
				self.assertEqual(lm_model.score_boundary     ,_lowerCamelCase )


	def 						_UpperCAmelCase     (self )							->							Optional[int]:







				'''simple docstring'''





				__lowercase										=							WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm''' )

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

				__lowercase										=							os.listdir(_lowerCamelCase )
				__lowercase										=							['''alphabet.json''', '''language_model''']

				downloaded_decoder_files.sort()
				expected_decoder_files.sort()

				# test that only decoder relevant files from
				# https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
				# are downloaded and none of the rest (e.g. README.md, ...)
				self.assertListEqual(_lowerCamelCase     ,_lowerCamelCase )


	def 						_UpperCAmelCase     (self )							->							List[Any]:







				'''simple docstring'''





				__lowercase										=							snapshot_download('''hf-internal-testing/processor_with_lm''' )

				__lowercase										=							WavaVecaProcessorWithLM.from_pretrained(_lowerCamelCase )

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

				__lowercase										=							os.listdir(_lowerCamelCase )
				__lowercase										=							os.listdir(_lowerCamelCase )

				local_decoder_files.sort()
				expected_decoder_files.sort()

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


	def 						_UpperCAmelCase     (self )							->							Dict:







				'''simple docstring'''





				__lowercase										=							WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm''' )
				__lowercase										=							AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm''' )

				__lowercase										=							floats_list((3, 1000) )

				__lowercase										=							processor_wavaveca(_lowerCamelCase     ,return_tensors='''np''' )
				__lowercase										=							processor_auto(_lowerCamelCase     ,return_tensors='''np''' )

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

				__lowercase										=							self._get_dummy_logits()

				__lowercase										=							processor_wavaveca.batch_decode(_lowerCamelCase )
				__lowercase										=							processor_auto.batch_decode(_lowerCamelCase )

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


	def 						_UpperCAmelCase     (self )							->							Dict:







				'''simple docstring'''





				__lowercase										=							self.get_feature_extractor()
				__lowercase										=							self.get_tokenizer()
				__lowercase										=							self.get_decoder()

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

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


	@staticmethod
	def 						_UpperCAmelCase     (_lowerCamelCase     ,_lowerCamelCase )							->							Optional[Any]:







				'''simple docstring'''





				__lowercase										=							[d[key] for d in offsets]
				return retrieved_list


	def 						_UpperCAmelCase     (self )							->							Optional[int]:







				'''simple docstring'''





				__lowercase										=							WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm''' )
				__lowercase										=							self._get_dummy_logits()[0]

				__lowercase										=							processor.decode(_lowerCamelCase     ,output_word_offsets=_lowerCamelCase )
				# check Wav2Vec2CTCTokenizerOutput keys for word
				self.assertEqual(len(outputs.keys() )     ,4 )
				self.assertTrue('''text''' in outputs )
				self.assertTrue('''word_offsets''' in outputs )
				self.assertTrue(isinstance(_lowerCamelCase     ,_lowerCamelCase ) )

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


	def 						_UpperCAmelCase     (self )							->							Optional[Any]:







				'''simple docstring'''





				__lowercase										=							WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm''' )
				__lowercase										=							self._get_dummy_logits()

				__lowercase										=							processor.batch_decode(_lowerCamelCase     ,output_word_offsets=_lowerCamelCase )

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

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


	@slow
	@require_torch
	@require_torchaudio
	def 						_UpperCAmelCase     (self )							->							List[Any]:







				'''simple docstring'''





				import torch

				__lowercase										=							load_dataset('''common_voice'''     ,'''en'''     ,split='''train'''     ,streaming=_lowerCamelCase )
				__lowercase										=							ds.cast_column('''audio'''     ,datasets.Audio(sampling_rate=16000 ) )
				__lowercase										=							iter(_lowerCamelCase )
				__lowercase										=							next(_lowerCamelCase )

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

				with torch.no_grad():
							__lowercase										=							model(_lowerCamelCase ).logits.cpu().numpy()

				__lowercase										=							processor.decode(logits[0]     ,output_word_offsets=_lowerCamelCase )

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

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

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

				# output times
				__lowercase										=							torch.tensor(self.get_from_offsets(_lowerCamelCase     ,'''start_time''' ) )
				__lowercase										=							torch.tensor(self.get_from_offsets(_lowerCamelCase     ,'''end_time''' ) )

				# fmt: off
				__lowercase										=							torch.tensor([1.4_1_9_9, 1.6_5_9_9, 2.2_5_9_9, 3.0, 3.2_4, 3.5_9_9_9, 3.7_9_9_9, 4.0_9_9_9, 4.2_6, 4.9_4, 5.2_8, 5.6_5_9_9, 5.7_8, 5.9_4, 6.3_2, 6.5_3_9_9, 6.6_5_9_9] )
				__lowercase										=							torch.tensor([1.5_3_9_9, 1.8_9_9_9, 2.9, 3.1_6, 3.5_3_9_9, 3.7_2, 4.0_1_9_9, 4.1_7_9_9, 4.7_6, 5.1_5_9_9, 5.5_5_9_9, 5.6_9_9_9, 5.8_6, 6.1_9_9_9, 6.3_8, 6.6_1_9_9, 6.9_4] )
				# fmt: on

				self.assertTrue(torch.allclose(_lowerCamelCase     ,_lowerCamelCase     ,atol=0.0_1 ) )
				self.assertTrue(torch.allclose(_lowerCamelCase     ,_lowerCamelCase     ,atol=0.0_1 ) )





