



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




					def 		lowerCAmelCase     (						self					)						->							Optional[int]:
										_snake_case													=			'| <pad> <unk> <s> </s> a b c d e f g h i j k'.split()
										_snake_case													=			dict(zip(lowerCAmelCase_			,  range(len(lowerCAmelCase_					)					)					)					)

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

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




					def 		lowerCAmelCase     (						self			,  **lowerCAmelCase_					)						->							Union[str, Any]:
										_snake_case													=			self.add_kwargs_tokens_map.copy()
										kwargs.update(lowerCAmelCase_					)
										return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname			,  **lowerCAmelCase_					)




					def 		lowerCAmelCase     (						self			,  **lowerCAmelCase_					)						->							List[str]:
										return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname			,  **lowerCAmelCase_					)




					def 		lowerCAmelCase     (						self			,  **lowerCAmelCase_					)						->							Union[str, Any]:
										return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name			,  **lowerCAmelCase_					)




					def 		lowerCAmelCase     (						self					)						->							str:
										shutil.rmtree(self.tmpdirname					)




					def 		lowerCAmelCase     (						self					)						->							List[Any]:
										_snake_case													=			self.get_tokenizer()
										_snake_case													=			self.get_feature_extractor()
										_snake_case													=			self.get_decoder()

										_snake_case													=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,  feature_extractor=lowerCAmelCase_			,  decoder=lowerCAmelCase_					)

										processor.save_pretrained(self.tmpdirname					)
										_snake_case													=			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 		lowerCAmelCase     (						self					)						->							Optional[Any]:
										_snake_case													=			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													=			WavaVecaProcessorWithLM.from_pretrained(
										    self.tmpdirname			,  alpha=5.0			,  beta=3.0			,  score_boundary=-7.0			,  unk_score_offset=3					)

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




					def 		lowerCAmelCase     (						self					)						->							Tuple:
										_snake_case													=			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 		lowerCAmelCase     (						self					)						->							Optional[int]:
										_snake_case													=			self.get_feature_extractor()
										_snake_case													=			self.get_tokenizer()
										_snake_case													=			self.get_decoder()

										_snake_case													=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,  feature_extractor=lowerCAmelCase_			,  decoder=lowerCAmelCase_					)

										_snake_case													=			floats_list((3, 1000)					)

										_snake_case													=			feature_extractor(lowerCAmelCase_			,  return_tensors='np'					)
										_snake_case													=			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 		lowerCAmelCase     (						self					)						->							Tuple:
										_snake_case													=			self.get_feature_extractor()
										_snake_case													=			self.get_tokenizer()
										_snake_case													=			self.get_decoder()

										_snake_case													=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,  feature_extractor=lowerCAmelCase_			,  decoder=lowerCAmelCase_					)

										_snake_case													=			'This is a test string'

										_snake_case													=			processor(text=lowerCAmelCase_					)

										_snake_case													=			tokenizer(lowerCAmelCase_					)

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




					def 		lowerCAmelCase     (						self			,  lowerCAmelCase_=(2, 10, 16)			,  lowerCAmelCase_=77					)						->							List[Any]:
										np.random.seed(lowerCAmelCase_					)
										return np.random.rand(*lowerCAmelCase_					)




					def 		lowerCAmelCase     (						self					)						->							str:
										_snake_case													=			self.get_feature_extractor()
										_snake_case													=			self.get_tokenizer()
										_snake_case													=			self.get_decoder()

										_snake_case													=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,  feature_extractor=lowerCAmelCase_			,  decoder=lowerCAmelCase_					)

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

										_snake_case													=			processor.decode(lowerCAmelCase_					)

										_snake_case													=			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 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							Optional[int]:
										_snake_case													=			self.get_feature_extractor()
										_snake_case													=			self.get_tokenizer()
										_snake_case													=			self.get_decoder()

										_snake_case													=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,  feature_extractor=lowerCAmelCase_			,  decoder=lowerCAmelCase_					)

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

										_snake_case													=			list(lowerCAmelCase_					)

										with get_context('fork'					).Pool() as p:
															_snake_case													=			decoder.decode_beams_batch(lowerCAmelCase_			,  lowerCAmelCase_					)

										_snake_case							,       _snake_case							,       _snake_case													=			[], [], []
										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 		lowerCAmelCase     (						self					)						->							List[Any]:
										_snake_case													=			self.get_feature_extractor()
										_snake_case													=			self.get_tokenizer()
										_snake_case													=			self.get_decoder()

										_snake_case													=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,  feature_extractor=lowerCAmelCase_			,  decoder=lowerCAmelCase_					)

										_snake_case													=			self._get_dummy_logits()

										_snake_case													=			15
										_snake_case													=			-20.0
										_snake_case													=			-4.0

										_snake_case													=			processor.batch_decode(
										    lowerCAmelCase_			,  beam_width=lowerCAmelCase_			,  beam_prune_logp=lowerCAmelCase_			,  token_min_logp=lowerCAmelCase_			,  )
										_snake_case													=			decoded_processor_out.text

										_snake_case													=			list(lowerCAmelCase_					)

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

										_snake_case													=			[d[0][0] for d in decoded_decoder_out]
										_snake_case													=			[d[0][2] for d in decoded_decoder_out]
										_snake_case													=			[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([-20.0_54, -18.4_47]			,  lowerCAmelCase_			,  atol=1E-3					)					)

										self.assertTrue(np.array_equal(lowerCAmelCase_			,  decoded_processor_out.lm_score					)					)
										self.assertTrue(np.allclose([-15.5_54, -13.94_74]			,  lowerCAmelCase_			,  atol=1E-3					)					)




					def 		lowerCAmelCase     (						self					)						->							Optional[Any]:
										_snake_case													=			self.get_feature_extractor()
										_snake_case													=			self.get_tokenizer()
										_snake_case													=			self.get_decoder()

										_snake_case													=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,  feature_extractor=lowerCAmelCase_			,  decoder=lowerCAmelCase_					)

										_snake_case													=			self._get_dummy_logits()

										_snake_case													=			2.0
										_snake_case													=			5.0
										_snake_case													=			-20.0
										_snake_case													=			True

										_snake_case													=			processor.batch_decode(
										    lowerCAmelCase_			,  alpha=lowerCAmelCase_			,  beta=lowerCAmelCase_			,  unk_score_offset=lowerCAmelCase_			,  lm_score_boundary=lowerCAmelCase_			,  )
										_snake_case													=			decoded_processor_out.text

										_snake_case													=			list(lowerCAmelCase_					)
										decoder.reset_params(
										    alpha=lowerCAmelCase_			,  beta=lowerCAmelCase_			,  unk_score_offset=lowerCAmelCase_			,  lm_score_boundary=lowerCAmelCase_			,  )

										with get_context('fork'					).Pool() as pool:
															_snake_case													=			decoder.decode_beams_batch(
															    lowerCAmelCase_			,  lowerCAmelCase_			,  )

										_snake_case													=			[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_					)
										_snake_case													=			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			,  lowerCAmelCase_					)




					def 		lowerCAmelCase     (						self					)						->							Any:
										_snake_case													=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'					)

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

										_snake_case													=			os.listdir(lowerCAmelCase_					)
										_snake_case													=			['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 		lowerCAmelCase     (						self					)						->							Optional[int]:
										_snake_case													=			snapshot_download('hf-internal-testing/processor_with_lm'					)

										_snake_case													=			WavaVecaProcessorWithLM.from_pretrained(lowerCAmelCase_					)

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

										_snake_case													=			os.listdir(lowerCAmelCase_					)
										_snake_case													=			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 		lowerCAmelCase     (						self					)						->							str:
										_snake_case													=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'					)
										_snake_case													=			AutoProcessor.from_pretrained('hf-internal-testing/processor_with_lm'					)

										_snake_case													=			floats_list((3, 1000)					)

										_snake_case													=			processor_wavaveca(lowerCAmelCase_			,  return_tensors='np'					)
										_snake_case													=			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					)

										_snake_case													=			self._get_dummy_logits()

										_snake_case													=			processor_wavaveca.batch_decode(lowerCAmelCase_					)
										_snake_case													=			processor_auto.batch_decode(lowerCAmelCase_					)

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




					def 		lowerCAmelCase     (						self					)						->							Tuple:
										_snake_case													=			self.get_feature_extractor()
										_snake_case													=			self.get_tokenizer()
										_snake_case													=			self.get_decoder()

										_snake_case													=			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 		lowerCAmelCase     (						lowerCAmelCase_			,  lowerCAmelCase_					)						->							Dict:
										_snake_case													=			[d[key] for d in offsets]
										return retrieved_list




					def 		lowerCAmelCase     (						self					)						->							List[Any]:
										_snake_case													=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'					)
										_snake_case													=			self._get_dummy_logits()[0]

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

										_snake_case													=			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 		lowerCAmelCase     (						self					)						->							int:
										import torch

										_snake_case													=			load_dataset('common_voice'			,  'en'			,  split='train'			,  streaming=lowerCAmelCase_					)
										_snake_case													=			ds.cast_column('audio'			,  datasets.Audio(sampling_rate=1_6000					)					)
										_snake_case													=			iter(lowerCAmelCase_					)
										_snake_case													=			next(lowerCAmelCase_					)

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

										with torch.no_grad():
															_snake_case													=			model(lowerCAmelCase_					).logits.cpu().numpy()

										_snake_case													=			processor.decode(logits[0]			,  output_word_offsets=lowerCAmelCase_					)

										_snake_case													=			model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
										_snake_case													=			[
										    {
										        '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													=			'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
										_snake_case													=			torch.tensor(self.get_from_offsets(lowerCAmelCase_			,  'start_time'					)					)
										_snake_case													=			torch.tensor(self.get_from_offsets(lowerCAmelCase_			,  'end_time'					)					)

										# fmt: off
										_snake_case													=			torch.tensor([1.41_99, 1.65_99, 2.25_99, 3.0, 3.24, 3.59_99, 3.79_99, 4.09_99, 4.26, 4.94, 5.28, 5.65_99, 5.78, 5.94, 6.32, 6.53_99, 6.65_99]					)
										_snake_case													=			torch.tensor([1.53_99, 1.89_99, 2.9, 3.16, 3.53_99, 3.72, 4.01_99, 4.17_99, 4.76, 5.15_99, 5.55_99, 5.69_99, 5.86, 6.19_99, 6.38, 6.61_99, 6.94]					)
										# fmt: on

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










def   lowerCamelCase__       (       UpperCamelCase__     :    Tuple    ,      UpperCamelCase__     :    Optional[int]  )     ->	Tuple:



					'''simple docstring'''




					_snake_case													=			[0 for i in range(r + 1  )]
					# nc0 = 1
					_snake_case													=			1
					for i in range(1    ,      n + 1  ):
										# to compute current row from previous row.
										_snake_case													=			min(UpperCamelCase__    ,      UpperCamelCase__  )
										while j > 0:
															c[j] += c[j - 1]
															j -= 1
					return c[r]




print(binomial_coefficient(n=10, r=5))



