



"""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        lowerCamelCase_   (  self						)			->							Optional[Any]:
									_UpperCAmelCase							 =		'| <pad> <unk> <s> </s> a b c d e f g h i j k'.split()
									_UpperCAmelCase							 =		dict(zip(snake_case	,       range(len(snake_case						)						)						)						)

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

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

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

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






							def        lowerCamelCase_   (  self	,       **snake_case						)			->							Tuple:
									_UpperCAmelCase							 =		self.add_kwargs_tokens_map.copy()
									kwargs.update(snake_case						)
									return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname	,       **snake_case						)






							def        lowerCamelCase_   (  self	,       **snake_case						)			->							int:
									return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname	,       **snake_case						)






							def        lowerCamelCase_   (  self	,       **snake_case						)			->							int:
									return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name	,       **snake_case						)






							def        lowerCamelCase_   (  self						)			->							Union[str, Any]:
									shutil.rmtree(self.tmpdirname						)






							def        lowerCamelCase_   (  self						)			->							List[str]:
									_UpperCAmelCase							 =		self.get_tokenizer()
									_UpperCAmelCase							 =		self.get_feature_extractor()
									_UpperCAmelCase							 =		self.get_decoder()

									_UpperCAmelCase							 =		WavaVecaProcessorWithLM(tokenizer=snake_case	,       feature_extractor=snake_case	,       decoder=snake_case						)

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

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

									# feature extractor
									self.assertEqual(processor.feature_extractor.to_json_string()	,       feature_extractor.to_json_string()						)
									self.assertIsInstance(processor.feature_extractor	,       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	,       snake_case						)






							def        lowerCamelCase_   (  self						)			->							List[str]:
									_UpperCAmelCase							 =		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
									_UpperCAmelCase							 =		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						)			->							int:
									_UpperCAmelCase							 =		self.get_tokenizer()
									# add token to trigger raise
									tokenizer.add_tokens(['xx']						)
									with self.assertRaisesRegex(snake_case	,       'include'						):
											WavaVecaProcessorWithLM(
											    tokenizer=snake_case	,       feature_extractor=self.get_feature_extractor()	,       decoder=self.get_decoder()						)






							def        lowerCamelCase_   (  self						)			->							Optional[int]:
									_UpperCAmelCase							 =		self.get_feature_extractor()
									_UpperCAmelCase							 =		self.get_tokenizer()
									_UpperCAmelCase							 =		self.get_decoder()

									_UpperCAmelCase							 =		WavaVecaProcessorWithLM(tokenizer=snake_case	,       feature_extractor=snake_case	,       decoder=snake_case						)

									_UpperCAmelCase							 =		floats_list((3, 1000)						)

									_UpperCAmelCase							 =		feature_extractor(snake_case	,       return_tensors='np'						)
									_UpperCAmelCase							 =		processor(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        lowerCamelCase_   (  self						)			->							int:
									_UpperCAmelCase							 =		self.get_feature_extractor()
									_UpperCAmelCase							 =		self.get_tokenizer()
									_UpperCAmelCase							 =		self.get_decoder()

									_UpperCAmelCase							 =		WavaVecaProcessorWithLM(tokenizer=snake_case	,       feature_extractor=snake_case	,       decoder=snake_case						)

									_UpperCAmelCase							 =		'This is a test string'

									_UpperCAmelCase							 =		processor(text=snake_case						)

									_UpperCAmelCase							 =		tokenizer(snake_case						)

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






							def        lowerCamelCase_   (  self	,       snake_case=(2, 10, 16)	,       snake_case=77						)			->							Optional[Any]:
									np.random.seed(snake_case						)
									return np.random.rand(*snake_case						)






							def        lowerCamelCase_   (  self						)			->							Optional[int]:
									_UpperCAmelCase							 =		self.get_feature_extractor()
									_UpperCAmelCase							 =		self.get_tokenizer()
									_UpperCAmelCase							 =		self.get_decoder()

									_UpperCAmelCase							 =		WavaVecaProcessorWithLM(tokenizer=snake_case	,       feature_extractor=snake_case	,       decoder=snake_case						)

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

									_UpperCAmelCase							 =		processor.decode(snake_case						)

									_UpperCAmelCase							 =		decoder.decode_beams(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        lowerCamelCase_   (  self	,       snake_case						)			->							List[Any]:
									_UpperCAmelCase							 =		self.get_feature_extractor()
									_UpperCAmelCase							 =		self.get_tokenizer()
									_UpperCAmelCase							 =		self.get_decoder()

									_UpperCAmelCase							 =		WavaVecaProcessorWithLM(tokenizer=snake_case	,       feature_extractor=snake_case	,       decoder=snake_case						)

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

									_UpperCAmelCase							 =		list(snake_case						)

									with get_context('fork'						).Pool() as p:
											_UpperCAmelCase							 =		decoder.decode_beams_batch(snake_case	,       snake_case						)

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






							def        lowerCamelCase_   (  self						)			->							Optional[int]:
									_UpperCAmelCase							 =		self.get_feature_extractor()
									_UpperCAmelCase							 =		self.get_tokenizer()
									_UpperCAmelCase							 =		self.get_decoder()

									_UpperCAmelCase							 =		WavaVecaProcessorWithLM(tokenizer=snake_case	,       feature_extractor=snake_case	,       decoder=snake_case						)

									_UpperCAmelCase							 =		self._get_dummy_logits()

									_UpperCAmelCase							 =		15
									_UpperCAmelCase							 =		-20.0
									_UpperCAmelCase							 =		-4.0

									_UpperCAmelCase							 =		processor.batch_decode(
									    snake_case	,       beam_width=snake_case	,       beam_prune_logp=snake_case	,       token_min_logp=snake_case	,       )
									_UpperCAmelCase							 =		decoded_processor_out.text

									_UpperCAmelCase							 =		list(snake_case						)

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

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

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

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

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






							def        lowerCamelCase_   (  self						)			->							int:
									_UpperCAmelCase							 =		self.get_feature_extractor()
									_UpperCAmelCase							 =		self.get_tokenizer()
									_UpperCAmelCase							 =		self.get_decoder()

									_UpperCAmelCase							 =		WavaVecaProcessorWithLM(tokenizer=snake_case	,       feature_extractor=snake_case	,       decoder=snake_case						)

									_UpperCAmelCase							 =		self._get_dummy_logits()

									_UpperCAmelCase							 =		2.0
									_UpperCAmelCase							 =		5.0
									_UpperCAmelCase							 =		-20.0
									_UpperCAmelCase							 =		True

									_UpperCAmelCase							 =		processor.batch_decode(
									    snake_case	,       alpha=snake_case	,       beta=snake_case	,       unk_score_offset=snake_case	,       lm_score_boundary=snake_case	,       )
									_UpperCAmelCase							 =		decoded_processor_out.text

									_UpperCAmelCase							 =		list(snake_case						)
									decoder.reset_params(
									    alpha=snake_case	,       beta=snake_case	,       unk_score_offset=snake_case	,       lm_score_boundary=snake_case	,       )

									with get_context('fork'						).Pool() as pool:
											_UpperCAmelCase							 =		decoder.decode_beams_batch(
											    snake_case	,       snake_case	,       )

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

									self.assertListEqual(snake_case	,       snake_case						)
									self.assertListEqual(['<s> </s> <s> </s> </s>', '</s> </s> <s> </s> </s>']	,       snake_case						)
									_UpperCAmelCase							 =		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	,       snake_case						)






							def        lowerCamelCase_   (  self						)			->							str:
									_UpperCAmelCase							 =		WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'						)

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

									_UpperCAmelCase							 =		os.listdir(snake_case						)
									_UpperCAmelCase							 =		['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(snake_case	,       snake_case						)






							def        lowerCamelCase_   (  self						)			->							Optional[int]:
									_UpperCAmelCase							 =		snapshot_download('hf-internal-testing/processor_with_lm'						)

									_UpperCAmelCase							 =		WavaVecaProcessorWithLM.from_pretrained(snake_case						)

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

									_UpperCAmelCase							 =		os.listdir(snake_case						)
									_UpperCAmelCase							 =		os.listdir(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(snake_case	,       snake_case						)






							def        lowerCamelCase_   (  self						)			->							Union[str, Any]:
									_UpperCAmelCase							 =		WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'						)
									_UpperCAmelCase							 =		AutoProcessor.from_pretrained('hf-internal-testing/processor_with_lm'						)

									_UpperCAmelCase							 =		floats_list((3, 1000)						)

									_UpperCAmelCase							 =		processor_wavaveca(snake_case	,       return_tensors='np'						)
									_UpperCAmelCase							 =		processor_auto(snake_case	,       return_tensors='np'						)

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

									_UpperCAmelCase							 =		self._get_dummy_logits()

									_UpperCAmelCase							 =		processor_wavaveca.batch_decode(snake_case						)
									_UpperCAmelCase							 =		processor_auto.batch_decode(snake_case						)

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






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

									_UpperCAmelCase							 =		WavaVecaProcessorWithLM(tokenizer=snake_case	,       feature_extractor=snake_case	,       decoder=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        lowerCamelCase_   (  snake_case	,       snake_case						)			->							Dict:
									_UpperCAmelCase							 =		[d[key] for d in offsets]
									return retrieved_list






							def        lowerCamelCase_   (  self						)			->							Optional[int]:
									_UpperCAmelCase							 =		WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'						)
									_UpperCAmelCase							 =		self._get_dummy_logits()[0]

									_UpperCAmelCase							 =		processor.decode(snake_case	,       output_word_offsets=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(snake_case	,       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        lowerCamelCase_   (  self						)			->							Union[str, Any]:
									_UpperCAmelCase							 =		WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'						)
									_UpperCAmelCase							 =		self._get_dummy_logits()

									_UpperCAmelCase							 =		processor.batch_decode(snake_case	,       output_word_offsets=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(snake_case	,       snake_case						)						)

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

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

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

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

									_UpperCAmelCase							 =		processor.decode(logits[0]	,       output_word_offsets=snake_case						)

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

									_UpperCAmelCase							 =		'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(snake_case	,       'word'						)						)	,       snake_case						)
									self.assertEqual(' '.join(self.get_from_offsets(snake_case	,       'word'						)						)	,       output.text						)

									# output times
									_UpperCAmelCase							 =		torch.tensor(self.get_from_offsets(snake_case	,       'start_time'						)						)
									_UpperCAmelCase							 =		torch.tensor(self.get_from_offsets(snake_case	,       'end_time'						)						)

									# fmt: off
									_UpperCAmelCase							 =		torch.tensor([1.4199, 1.6599, 2.2599, 3.0, 3.24, 3.5999, 3.7999, 4.0999, 4.26, 4.94, 5.28, 5.6599, 5.78, 5.94, 6.32, 6.5399, 6.6599]						)
									_UpperCAmelCase							 =		torch.tensor([1.5399, 1.8999, 2.9, 3.16, 3.5399, 3.72, 4.0199, 4.1799, 4.76, 5.1599, 5.5599, 5.6999, 5.86, 6.1999, 6.38, 6.6199, 6.94]						)
									# fmt: on

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







"""simple docstring"""





import warnings

from ...utils import logging
from .image_processing_yolos import YolosImageProcessor


lowercase  							=					logging.get_logger(__name__)




class 			lowercase__			(      A    ):
							'''simple docstring'''







							def __init__(  self	,       *snake_case	,       **snake_case						)			->							None:
									warnings.warn(
									    'The class YolosFeatureExtractor is deprecated and will be removed in version 5 of Transformers. Please'
									    ' use YolosImageProcessor instead.'	,       snake_case	,       )
									super().__init__(*snake_case	,       **snake_case						)


