

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






					"""simple docstring"""



					def 			__lowerCAmelCase					(			self     :							List[str]						):
						lowerCAmelCase__      :		Optional[Any]    							=      '''| <pad> <unk> <s> </s> a b c d e f g h i j k'''.split()
						lowerCAmelCase__      :		List[Any]    							=      dict(zip(lowercase_  ,range(len(lowercase_						)						)						)						)

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

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

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

						# load decoder from hub
						lowerCAmelCase__      :		List[Any]    							=      '''hf-internal-testing/ngram-beam-search-decoder'''







					def 			__lowerCAmelCase					(			self     :							str  ,**lowercase_     :							Union[str, Any]						):
						lowerCAmelCase__      :		Any    							=      self.add_kwargs_tokens_map.copy()
						kwargs.update(lowercase_						)
						return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname  ,**lowercase_						)







					def 			__lowerCAmelCase					(			self     :							List[Any]  ,**lowercase_     :							Optional[int]						):
						return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname  ,**lowercase_						)







					def 			__lowerCAmelCase					(			self     :							Optional[int]  ,**lowercase_     :							Optional[Any]						):
						return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name  ,**lowercase_						)







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







					def 			__lowerCAmelCase					(			self     :							List[Any]						):
						lowerCAmelCase__      :		Dict    							=      self.get_tokenizer()
						lowerCAmelCase__      :		Tuple    							=      self.get_feature_extractor()
						lowerCAmelCase__      :		Dict    							=      self.get_decoder()

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

						processor.save_pretrained(self.tmpdirname						)
						lowerCAmelCase__      :		Union[str, Any]    							=      WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname						)

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

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

						# 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  ,lowercase_						)







					def 			__lowerCAmelCase					(			self     :							Optional[int]						):
						lowerCAmelCase__      :		List[str]    							=      WavaVecaProcessorWithLM(
						    tokenizer=self.get_tokenizer()  ,feature_extractor=self.get_feature_extractor()  ,decoder=self.get_decoder()						)
						processor.save_pretrained(self.tmpdirname						)

						# make sure that error is thrown when decoder alphabet doesn't match
						lowerCAmelCase__      :		str    							=      WavaVecaProcessorWithLM.from_pretrained(
						    self.tmpdirname  ,alpha=5.0  ,beta=3.0  ,score_boundary=-7.0  ,unk_score_offset=3						)

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







					def 			__lowerCAmelCase					(			self     :							int						):
						lowerCAmelCase__      :		str    							=      self.get_tokenizer()
						# add token to trigger raise
						tokenizer.add_tokens(['''xx''']						)
						with self.assertRaisesRegex(lowercase_  ,'''include'''						):
							WavaVecaProcessorWithLM(
							    tokenizer=lowercase_  ,feature_extractor=self.get_feature_extractor()  ,decoder=self.get_decoder()						)







					def 			__lowerCAmelCase					(			self     :							Dict						):
						lowerCAmelCase__      :		Union[str, Any]    							=      self.get_feature_extractor()
						lowerCAmelCase__      :		Union[str, Any]    							=      self.get_tokenizer()
						lowerCAmelCase__      :		Dict    							=      self.get_decoder()

						lowerCAmelCase__      :		List[str]    							=      WavaVecaProcessorWithLM(tokenizer=lowercase_  ,feature_extractor=lowercase_  ,decoder=lowercase_						)

						lowerCAmelCase__      :		Any    							=      floats_list((3, 1_0_0_0)						)

						lowerCAmelCase__      :		Dict    							=      feature_extractor(lowercase_  ,return_tensors='''np'''						)
						lowerCAmelCase__      :		str    							=      processor(lowercase_  ,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     :							List[Any]						):
						lowerCAmelCase__      :		Optional[int]    							=      self.get_feature_extractor()
						lowerCAmelCase__      :		Union[str, Any]    							=      self.get_tokenizer()
						lowerCAmelCase__      :		Optional[Any]    							=      self.get_decoder()

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

						lowerCAmelCase__      :		Dict    							=      '''This is a test string'''

						lowerCAmelCase__      :		Any    							=      processor(text=lowercase_						)

						lowerCAmelCase__      :		str    							=      tokenizer(lowercase_						)

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







					def 			__lowerCAmelCase					(			self     :							Optional[int]  ,lowercase_     :							Tuple=(2, 1_0, 1_6)  ,lowercase_     :							Optional[int]=7_7						):
						np.random.seed(lowercase_						)
						return np.random.rand(*lowercase_						)







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

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

						lowerCAmelCase__      :		Optional[Any]    							=      self._get_dummy_logits(shape=(1_0, 1_6)  ,seed=1_3						)

						lowerCAmelCase__      :		Union[str, Any]    							=      processor.decode(lowercase_						)

						lowerCAmelCase__      :		Union[str, Any]    							=      decoder.decode_beams(lowercase_						)[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     :							Union[str, Any]  ,lowercase_     :							Tuple						):
						lowerCAmelCase__      :		Tuple    							=      self.get_feature_extractor()
						lowerCAmelCase__      :		Optional[Any]    							=      self.get_tokenizer()
						lowerCAmelCase__      :		str    							=      self.get_decoder()

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

						lowerCAmelCase__      :		str    							=      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:
							lowerCAmelCase__      :		List[str]    							=      processor.batch_decode(lowercase_						)
						else:
							with get_context(lowercase_						).Pool() as pool:
								lowerCAmelCase__      :		str    							=      processor.batch_decode(lowercase_  ,lowercase_						)

						lowerCAmelCase__      :		Union[str, Any]    							=      list(lowercase_						)

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

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

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







					def 			__lowerCAmelCase					(			self     :							Any						):
						lowerCAmelCase__      :		str    							=      self.get_feature_extractor()
						lowerCAmelCase__      :		Tuple    							=      self.get_tokenizer()
						lowerCAmelCase__      :		Optional[Any]    							=      self.get_decoder()

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

						lowerCAmelCase__      :		str    							=      self._get_dummy_logits()

						lowerCAmelCase__      :		List[str]    							=      1_5
						lowerCAmelCase__      :		Optional[Any]    							=      -20.0
						lowerCAmelCase__      :		Union[str, Any]    							=      -4.0

						lowerCAmelCase__      :		Union[str, Any]    							=      processor.batch_decode(
						    lowercase_  ,beam_width=lowercase_  ,beam_prune_logp=lowercase_  ,token_min_logp=lowercase_  ,)
						lowerCAmelCase__      :		int    							=      decoded_processor_out.text

						lowerCAmelCase__      :		List[str]    							=      list(lowercase_						)

						with get_context('''fork'''						).Pool() as pool:
							lowerCAmelCase__      :		Union[str, Any]    							=      decoder.decode_beams_batch(
							    lowercase_  ,lowercase_  ,beam_width=lowercase_  ,beam_prune_logp=lowercase_  ,token_min_logp=lowercase_  ,)

						lowerCAmelCase__      :		Dict    							=      [d[0][0] for d in decoded_decoder_out]
						lowerCAmelCase__      :		Optional[int]    							=      [d[0][2] for d in decoded_decoder_out]
						lowerCAmelCase__      :		List[Any]    							=      [d[0][3] for d in decoded_decoder_out]

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

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

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







					def 			__lowerCAmelCase					(			self     :							Optional[Any]						):
						lowerCAmelCase__      :		Union[str, Any]    							=      self.get_feature_extractor()
						lowerCAmelCase__      :		int    							=      self.get_tokenizer()
						lowerCAmelCase__      :		Optional[int]    							=      self.get_decoder()

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

						lowerCAmelCase__      :		str    							=      self._get_dummy_logits()

						lowerCAmelCase__      :		Optional[int]    							=      2.0
						lowerCAmelCase__      :		str    							=      5.0
						lowerCAmelCase__      :		Union[str, Any]    							=      -20.0
						lowerCAmelCase__      :		Optional[int]    							=      True

						lowerCAmelCase__      :		str    							=      processor.batch_decode(
						    lowercase_  ,alpha=lowercase_  ,beta=lowercase_  ,unk_score_offset=lowercase_  ,lm_score_boundary=lowercase_  ,)
						lowerCAmelCase__      :		int    							=      decoded_processor_out.text

						lowerCAmelCase__      :		List[Any]    							=      list(lowercase_						)
						decoder.reset_params(
						    alpha=lowercase_  ,beta=lowercase_  ,unk_score_offset=lowercase_  ,lm_score_boundary=lowercase_  ,)

						with get_context('''fork'''						).Pool() as pool:
							lowerCAmelCase__      :		int    							=      decoder.decode_beams_batch(
							    lowercase_  ,lowercase_  ,)

						lowerCAmelCase__      :		Optional[Any]    							=      [d[0][0] for d in decoded_decoder_out]

						self.assertListEqual(lowercase_  ,lowercase_						)
						self.assertListEqual(['''<s> </s> <s> </s> </s>''', '''</s> </s> <s> </s> </s>''']  ,lowercase_						)
						lowerCAmelCase__      :		int    							=      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  ,lowercase_						)







					def 			__lowerCAmelCase					(			self     :							List[str]						):
						lowerCAmelCase__      :		Any    							=      WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''						)

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

						lowerCAmelCase__      :		Dict    							=      os.listdir(lowercase_						)
						lowerCAmelCase__      :		List[str]    							=      ['''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(lowercase_  ,lowercase_						)







					def 			__lowerCAmelCase					(			self     :							Union[str, Any]						):
						lowerCAmelCase__      :		Union[str, Any]    							=      snapshot_download('''hf-internal-testing/processor_with_lm'''						)

						lowerCAmelCase__      :		Tuple    							=      WavaVecaProcessorWithLM.from_pretrained(lowercase_						)

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

						lowerCAmelCase__      :		Union[str, Any]    							=      os.listdir(lowercase_						)
						lowerCAmelCase__      :		Optional[int]    							=      os.listdir(lowercase_						)

						local_decoder_files.sort()
						expected_decoder_files.sort()

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







					def 			__lowerCAmelCase					(			self     :							Any						):
						lowerCAmelCase__      :		Optional[int]    							=      WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''						)
						lowerCAmelCase__      :		List[Any]    							=      AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''						)

						lowerCAmelCase__      :		Tuple    							=      floats_list((3, 1_0_0_0)						)

						lowerCAmelCase__      :		str    							=      processor_wavaveca(lowercase_  ,return_tensors='''np'''						)
						lowerCAmelCase__      :		Optional[Any]    							=      processor_auto(lowercase_  ,return_tensors='''np'''						)

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

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

						lowerCAmelCase__      :		Optional[int]    							=      processor_wavaveca.batch_decode(lowercase_						)
						lowerCAmelCase__      :		Tuple    							=      processor_auto.batch_decode(lowercase_						)

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







					def 			__lowerCAmelCase					(			self     :							Union[str, Any]						):
						lowerCAmelCase__      :		Union[str, Any]    							=      self.get_feature_extractor()
						lowerCAmelCase__      :		str    							=      self.get_tokenizer()
						lowerCAmelCase__      :		Optional[int]    							=      self.get_decoder()

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

						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					(			lowercase_     :							Dict  ,lowercase_     :							List[Any]						):
						lowerCAmelCase__      :		Tuple    							=      [d[key] for d in offsets]
						return retrieved_list







					def 			__lowerCAmelCase					(			self     :							Tuple						):
						lowerCAmelCase__      :		List[str]    							=      WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''						)
						lowerCAmelCase__      :		Any    							=      self._get_dummy_logits()[0]

						lowerCAmelCase__      :		List[str]    							=      processor.decode(lowercase_  ,output_word_offsets=lowercase_						)
						# 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(lowercase_  ,lowercase_						)						)

						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[int]						):
						lowerCAmelCase__      :		str    							=      WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''						)
						lowerCAmelCase__      :		Union[str, Any]    							=      self._get_dummy_logits()

						lowerCAmelCase__      :		Any    							=      processor.batch_decode(lowercase_  ,output_word_offsets=lowercase_						)

						# 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(lowercase_  ,lowercase_						)						)

						self.assertListEqual(
						    [''' '''.join(self.get_from_offsets(lowercase_  ,'''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

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

						lowerCAmelCase__      :		Union[str, Any]    							=      AutoProcessor.from_pretrained('''patrickvonplaten/wav2vec2-base-100h-with-lm'''						)
						lowerCAmelCase__      :		int    							=      WavaVecaForCTC.from_pretrained('''patrickvonplaten/wav2vec2-base-100h-with-lm'''						)

						# compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
						lowerCAmelCase__      :		Any    							=      processor(sample['''audio''']['''array''']  ,return_tensors='''pt'''						).input_values

						with torch.no_grad():
							lowerCAmelCase__      :		List[Any]    							=      model(lowercase_						).logits.cpu().numpy()

						lowerCAmelCase__      :		str    							=      processor.decode(logits[0]  ,output_word_offsets=lowercase_						)

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

						lowerCAmelCase__      :		Dict    							=      '''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(lowercase_  ,'''word'''						)						)  ,lowercase_						)
						self.assertEqual(''' '''.join(self.get_from_offsets(lowercase_  ,'''word'''						)						)  ,output.text						)

						# output times
						lowerCAmelCase__      :		Union[str, Any]    							=      torch.tensor(self.get_from_offsets(lowercase_  ,'''start_time'''						)						)
						lowerCAmelCase__      :		Optional[int]    							=      torch.tensor(self.get_from_offsets(lowercase_  ,'''end_time'''						)						)

						# fmt: off
						lowerCAmelCase__      :		Optional[Any]    							=      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]						)
						lowerCAmelCase__      :		List[str]    							=      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(lowercase_  ,lowercase_  ,atol=0.01						)						)
						self.assertTrue(torch.allclose(lowercase_  ,lowercase_  ,atol=0.01						)						)







"""simple docstring"""





from __future__ import annotations

import unittest

import numpy as np

from transformers import OPTConfig, is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
							import tensorflow as tf

							from transformers import GPTaTokenizer, TFOPTForCausalLM, TFOPTModel

def     __SCREAMING_SNAKE_CASE    ( A_		,	A_		,	A_=None		,	A_=None					):
	if attention_mask is None:
		lowerCAmelCase__      :		Optional[int]    							=      tf.cast(tf.math.not_equal(A_		,	config.pad_token_id					)		,	tf.inta					)
	return {"input_ids": input_ids, "attention_mask": attention_mask}







@require_tf
class 							SCREAMING_SNAKE_CASE   :






					"""simple docstring"""



					lowercase__												=			OPTConfig
					lowercase__												=			{}
					lowercase__												=			"gelu"







					def __init__(			self     :							str  ,lowercase_     :							List[Any]  ,lowercase_     :							Optional[int]=1_3  ,lowercase_     :							str=7  ,lowercase_     :							int=True  ,lowercase_     :							List[str]=False  ,lowercase_     :							Optional[int]=9_9  ,lowercase_     :							Union[str, Any]=1_6  ,lowercase_     :							Any=2  ,lowercase_     :							Tuple=4  ,lowercase_     :							Union[str, Any]=4  ,lowercase_     :							int="gelu"  ,lowercase_     :							Tuple=0.1  ,lowercase_     :							Optional[Any]=0.1  ,lowercase_     :							List[str]=2_0  ,lowercase_     :							Union[str, Any]=2  ,lowercase_     :							Union[str, Any]=1  ,lowercase_     :							Tuple=0  ,lowercase_     :							List[Any]=1_6  ,lowercase_     :							Union[str, Any]=1_6  ,):
						lowerCAmelCase__      :		Dict    							=      parent
						lowerCAmelCase__      :		Tuple    							=      batch_size
						lowerCAmelCase__      :		Dict    							=      seq_length
						lowerCAmelCase__      :		Optional[Any]    							=      is_training
						lowerCAmelCase__      :		Optional[Any]    							=      use_labels
						lowerCAmelCase__      :		Union[str, Any]    							=      vocab_size
						lowerCAmelCase__      :		Optional[int]    							=      hidden_size
						lowerCAmelCase__      :		int    							=      num_hidden_layers
						lowerCAmelCase__      :		Dict    							=      num_attention_heads
						lowerCAmelCase__      :		Union[str, Any]    							=      intermediate_size
						lowerCAmelCase__      :		Union[str, Any]    							=      hidden_act
						lowerCAmelCase__      :		int    							=      hidden_dropout_prob
						lowerCAmelCase__      :		Any    							=      attention_probs_dropout_prob
						lowerCAmelCase__      :		int    							=      max_position_embeddings
						lowerCAmelCase__      :		Any    							=      eos_token_id
						lowerCAmelCase__      :		List[Any]    							=      pad_token_id
						lowerCAmelCase__      :		Any    							=      bos_token_id
						lowerCAmelCase__      :		Dict    							=      embed_dim
						lowerCAmelCase__      :		int    							=      word_embed_proj_dim
						lowerCAmelCase__      :		Union[str, Any]    							=      False







					def 			__lowerCAmelCase					(			self     :							int						):
						lowerCAmelCase__      :		Optional[int]    							=      ids_tensor([self.batch_size, self.seq_length - 1]  ,self.vocab_size						)
						lowerCAmelCase__      :		List[Any]    							=      tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size						)  ,1						)
						lowerCAmelCase__      :		int    							=      tf.concat([input_ids, eos_tensor]  ,axis=1						)

						lowerCAmelCase__      :		Union[str, Any]    							=      self.config_cls(
						    vocab_size=self.vocab_size  ,hidden_size=self.hidden_size  ,num_hidden_layers=self.num_hidden_layers  ,num_attention_heads=self.num_attention_heads  ,ffn_dim=self.intermediate_size  ,dropout=self.hidden_dropout_prob  ,attention_dropout=self.attention_probs_dropout_prob  ,max_position_embeddings=self.max_position_embeddings  ,eos_token_id=self.eos_token_id  ,bos_token_id=self.bos_token_id  ,pad_token_id=self.pad_token_id  ,embed_dim=self.embed_dim  ,word_embed_proj_dim=self.word_embed_proj_dim  ,is_encoder_decoder=lowercase_  ,**self.config_updates  ,)
						lowerCAmelCase__      :		Optional[Any]    							=      prepare_opt_inputs_dict(lowercase_  ,lowercase_						)
						return config, inputs_dict







					def 			__lowerCAmelCase					(			self     :							List[Any]  ,lowercase_     :							int  ,lowercase_     :							List[Any]						):
						lowerCAmelCase__      :		Tuple    							=      TFOPTModel(config=lowercase_						)
						lowerCAmelCase__      :		Tuple    							=      inputs_dict['''input_ids''']

						lowerCAmelCase__      :		Optional[int]    							=      input_ids[:1, :]
						lowerCAmelCase__      :		List[str]    							=      inputs_dict['''attention_mask'''][:1, :]
						lowerCAmelCase__      :		Any    							=      1

						# first forward pass
						lowerCAmelCase__      :		Dict    							=      model(lowercase_  ,attention_mask=lowercase_  ,use_cache=lowercase_						)

						lowerCAmelCase__							,lowerCAmelCase__      :		Any    							=      outputs.to_tuple()

						# create hypothetical next token and extent to next_input_ids
						lowerCAmelCase__      :		List[Any]    							=      ids_tensor((self.batch_size, 3)  ,config.vocab_size						)
						lowerCAmelCase__      :		Tuple    							=      tf.cast(ids_tensor((self.batch_size, 3)  ,2						)  ,tf.inta						)

						# append to next input_ids and
						lowerCAmelCase__      :		Optional[Any]    							=      tf.concat([input_ids, next_tokens]  ,axis=-1						)
						lowerCAmelCase__      :		Tuple    							=      tf.concat([attention_mask, next_attn_mask]  ,axis=-1						)

						lowerCAmelCase__      :		Any    							=      model(lowercase_  ,attention_mask=lowercase_						)[0]
						lowerCAmelCase__      :		Any    							=      model(lowercase_  ,attention_mask=lowercase_  ,past_key_values=lowercase_						)[0]

						self.parent.assertEqual(next_tokens.shape[1]  ,output_from_past.shape[1]						)

						# select random slice
						lowerCAmelCase__      :		str    							=      int(ids_tensor((1,)  ,output_from_past.shape[-1]						)						)
						lowerCAmelCase__      :		List[str]    							=      output_from_no_past[:, -3:, random_slice_idx]
						lowerCAmelCase__      :		Union[str, Any]    							=      output_from_past[:, :, random_slice_idx]

						# test that outputs are equal for slice
						tf.debugging.assert_near(lowercase_  ,lowercase_  ,rtol=1E-3						)







@require_tf
class 							SCREAMING_SNAKE_CASE   (						a_       ,				a_       ,				unittest.TestCase			):






					"""simple docstring"""



					lowercase__												=			(TFOPTModel, TFOPTForCausalLM) if is_tf_available() else ()
					lowercase__												=			(TFOPTForCausalLM,) if is_tf_available() else ()
					lowercase__												=			(
					    {"feature-extraction": TFOPTModel, "text-generation": TFOPTForCausalLM} if is_tf_available() else {}
					)
					lowercase__												=			False
					lowercase__												=			False
					lowercase__												=			False
					lowercase__												=			10







					def 			__lowerCAmelCase					(			self     :							Any						):
						lowerCAmelCase__      :		List[Any]    							=      TFOPTModelTester(self						)
						lowerCAmelCase__      :		Optional[Any]    							=      ConfigTester(self  ,config_class=lowercase_						)







					def 			__lowerCAmelCase					(			self     :							str						):
						self.config_tester.run_common_tests()







					def 			__lowerCAmelCase					(			self     :							Any						):
						lowerCAmelCase__      :		Any    							=      self.model_tester.prepare_config_and_inputs_for_common()
						self.model_tester.check_decoder_model_past_large_inputs(*lowercase_						)







					def 			__lowerCAmelCase					(			self     :							Tuple						):
						lowerCAmelCase__							,lowerCAmelCase__      :		List[Any]    							=      self.model_tester.prepare_config_and_inputs_for_common()

						def _get_word_embedding_weight(lowercase_     :							Optional[Any]  ,lowercase_     :							List[Any]						):
							if hasattr(lowercase_  ,'''weight'''						):
								return embedding_layer.weight
							else:
								# Here we build the word embeddings weights if not exists.
								# And then we retry to get the attribute once built.
								model.build()
								if hasattr(lowercase_  ,'''weight'''						):
									return embedding_layer.weight
								else:
									return None

						for model_class in self.all_model_classes:
							for size in [config.vocab_size - 1_0, config.vocab_size + 1_0]:
								# build the embeddings
								lowerCAmelCase__      :		Optional[int]    							=      model_class(config=lowercase_						)
								lowerCAmelCase__      :		List[Any]    							=      _get_word_embedding_weight(lowercase_  ,model.get_input_embeddings()						)
								lowerCAmelCase__      :		Optional[Any]    							=      _get_word_embedding_weight(lowercase_  ,model.get_output_embeddings()						)

								# reshape the embeddings
								model.resize_token_embeddings(lowercase_						)
								lowerCAmelCase__      :		Union[str, Any]    							=      _get_word_embedding_weight(lowercase_  ,model.get_input_embeddings()						)
								lowerCAmelCase__      :		List[Any]    							=      _get_word_embedding_weight(lowercase_  ,model.get_output_embeddings()						)

								# check that the resized embeddings size matches the desired size.
								lowerCAmelCase__      :		Optional[Any]    							=      size if size is not None else config.vocab_size

								self.assertEqual(new_input_embeddings.shape[0]  ,lowercase_						)

								# check that weights remain the same after resizing
								lowerCAmelCase__      :		Optional[Any]    							=      True
								for pa, pa in zip(old_input_embeddings.value()  ,new_input_embeddings.value()						):
									if tf.math.reduce_sum(tf.math.abs(pa - pa						)						) > 0:
										lowerCAmelCase__      :		str    							=      False
								self.assertTrue(lowercase_						)

								if old_output_embeddings is not None and new_output_embeddings is not None:
									self.assertEqual(new_output_embeddings.shape[0]  ,lowercase_						)

									lowerCAmelCase__      :		Dict    							=      True
									for pa, pa in zip(old_output_embeddings.value()  ,new_output_embeddings.value()						):
										if tf.math.reduce_sum(tf.math.abs(pa - pa						)						) > 0:
											lowerCAmelCase__      :		List[Any]    							=      False
									self.assertTrue(lowercase_						)

def     __SCREAMING_SNAKE_CASE    ( A_					):
	return tf.constant(A_		,	dtype=tf.intaa					)







@require_tf
class 							SCREAMING_SNAKE_CASE   (						unittest.TestCase			):






					"""simple docstring"""



					lowercase__												=			99







					def 			__lowerCAmelCase					(			self     :							Any						):
						lowerCAmelCase__      :		Dict    							=      tf.ones((4, 1)  ,dtype=tf.intaa						) * 2
						lowerCAmelCase__      :		Any    							=      tf.concat([ids_tensor((4, 6)  ,self.vocab_size - 3						) + 3, eos_column_vector]  ,axis=1						)
						lowerCAmelCase__      :		str    							=      input_ids.shape[0]
						lowerCAmelCase__      :		str    							=      OPTConfig(
						    vocab_size=self.vocab_size  ,hidden_size=2_4  ,num_hidden_layers=2  ,num_attention_heads=2  ,ffn_dim=3_2  ,max_position_embeddings=4_8  ,eos_token_id=2  ,pad_token_id=1  ,bos_token_id=0  ,)
						return config, input_ids, batch_size







@require_sentencepiece
@require_tf
class 							SCREAMING_SNAKE_CASE   (						unittest.TestCase			):






					"""simple docstring"""



					@slow
					def 			__lowerCAmelCase					(			self     :							Optional[Any]						):
						lowerCAmelCase__      :		Tuple    							=      TFOPTModel.from_pretrained('''facebook/opt-350m'''						)
						lowerCAmelCase__      :		str    							=      _long_tensor([[0, 3_1_4_1_4, 2_3_2, 3_2_8, 7_4_0, 1_1_4_0, 1_2_6_9_5, 6_9, 4_6_0_7_8, 1_5_8_8, 2]]						)
						lowerCAmelCase__      :		List[Any]    							=      tf.not_equal(lowercase_  ,model.config.pad_token_id						)
						with tf.GradientTape():
							lowerCAmelCase__      :		List[str]    							=      model(input_ids=lowercase_  ,attention_mask=lowercase_						).last_hidden_state
						lowerCAmelCase__      :		Optional[int]    							=      (1, 1_1, 5_1_2)
						self.assertEqual(output.shape  ,lowercase_						)
						lowerCAmelCase__      :		Any    							=      tf.constant(
						    [[-0.2873, -1.9218, -0.3033], [-1.2710, -0.1338, -0.1902], [0.4095, 0.1214, -1.3121]]						)
						self.assertTrue(np.allclose(output[:, :3, :3]  ,lowercase_  ,atol=4E-3						)						)

						lowerCAmelCase__      :		Any    							=      tf.function(lowercase_  ,jit_compile=lowercase_						)
						lowerCAmelCase__      :		Optional[Any]    							=      xla_generate(lowercase_  ,lowercase_						)[0]
						self.assertTrue(np.allclose(output[:, :3, :3]  ,lowercase_  ,atol=4E-2						)						)







@require_tf
@slow
class 							SCREAMING_SNAKE_CASE   (						unittest.TestCase			):






					"""simple docstring"""



					def 			__lowerCAmelCase					(			self     :							Optional[Any]						):
						super().setUp()
						lowerCAmelCase__      :		Optional[int]    							=      '''facebook/opt-350m'''







					def 			__lowerCAmelCase					(			self     :							Any						):
						lowerCAmelCase__      :		Optional[int]    							=      TFOPTForCausalLM.from_pretrained(self.path_model						)
						lowerCAmelCase__      :		Union[str, Any]    							=      GPTaTokenizer.from_pretrained(self.path_model						)

						lowerCAmelCase__      :		int    							=      [
						    '''Today is a beautiful day and I want to''',
						    '''In the city of''',
						    '''Paris is the capital of France and''',
						    '''Computers and mobile phones have taken''',
						]
						# verify that prompt without BOS token is identical to Metaseq -> add_special_tokens=False
						lowerCAmelCase__      :		Optional[Any]    							=      tokenizer(lowercase_  ,return_tensors='''tf'''  ,padding=lowercase_  ,add_special_tokens=lowercase_						)
						lowerCAmelCase__      :		int    							=      tf.math.reduce_mean(model(inputs.input_ids  ,attention_mask=inputs.attention_mask						)[0]  ,axis=-1						)
						lowerCAmelCase__      :		List[str]    							=      tf.constant(
						    [
						        [1.3851, -13.8923, -10.5229, -10.7533, -0.2309, -10.2384, -0.5365, -9.0947, -5.1670],
						        [-4.7073, -10.6276, -3.9415, -21.5242, -0.2822, -0.2822, -0.2822, -0.2822, -0.2822],
						        [0.6247, -3.4229, -8.9179, -1.4297, -14.1650, 1.4146, -9.0218, -0.2703, -0.2703],
						        [6.4783, -1.9913, -10.7926, -2.3336, 1.5092, -0.9974, -6.8213, 1.3477, 1.3477],
						    ]						)
						self.assertTrue(np.allclose(lowercase_  ,lowercase_  ,atol=1E-4						)						)

						lowerCAmelCase__      :		int    							=      tf.function(lowercase_  ,jit_compile=lowercase_						)
						lowerCAmelCase__      :		Tuple    							=      tf.math.reduce_mean(xla_generate(inputs.input_ids  ,attention_mask=inputs.attention_mask						)[0]  ,axis=-1						)
						self.assertTrue(np.allclose(lowercase_  ,lowercase_  ,atol=1E-4						)						)







@require_tf
@slow
class 							SCREAMING_SNAKE_CASE   (						unittest.TestCase			):






					"""simple docstring"""



					@property
					def 			__lowerCAmelCase					(			self     :							List[str]						):
						return [
						    "Today is a beautiful day and I want",
						    "In the city of",
						    "Paris is the capital of France and",
						    "Computers and mobile phones have taken",
						]







					def 			__lowerCAmelCase					(			self     :							Any						):
						lowerCAmelCase__      :		Union[str, Any]    							=      '''facebook/opt-125m'''

						lowerCAmelCase__      :		str    							=      [
						    '''Today is a beautiful day and I want to''',
						    '''In the city of New York, the city''',
						    '''Paris is the capital of France and the capital''',
						    '''Computers and mobile phones have taken over the''',
						]

						lowerCAmelCase__      :		Any    							=      []
						lowerCAmelCase__      :		int    							=      GPTaTokenizer.from_pretrained(lowercase_						)
						lowerCAmelCase__      :		int    							=      TFOPTForCausalLM.from_pretrained(lowercase_						)

						for prompt in self.prompts:
							lowerCAmelCase__      :		Union[str, Any]    							=      tokenizer(lowercase_  ,return_tensors='''tf'''						).input_ids

							lowerCAmelCase__      :		Any    							=      model.generate(lowercase_  ,max_length=1_0						)

							lowerCAmelCase__      :		Optional[int]    							=      tokenizer.batch_decode(lowercase_  ,skip_special_tokens=lowercase_						)
							predicted_outputs += generated_string

						self.assertListEqual(lowercase_  ,lowercase_						)







					def 			__lowerCAmelCase					(			self     :							Any						):
						lowerCAmelCase__      :		Union[str, Any]    							=      '''facebook/opt-350m'''

						lowerCAmelCase__      :		Optional[Any]    							=      GPTaTokenizer.from_pretrained(lowercase_						)
						lowerCAmelCase__      :		Optional[Any]    							=      TFOPTForCausalLM.from_pretrained(lowercase_						)

						lowerCAmelCase__      :		Optional[Any]    							=      '''left'''

						# use different length sentences to test batching
						lowerCAmelCase__      :		Dict    							=      [
						    '''Hello, my dog is a little''',
						    '''Today, I''',
						]

						lowerCAmelCase__      :		Union[str, Any]    							=      tokenizer(lowercase_  ,return_tensors='''tf'''  ,padding=lowercase_						)
						lowerCAmelCase__      :		Optional[Any]    							=      inputs['''input_ids''']

						lowerCAmelCase__      :		str    							=      model.generate(input_ids=lowercase_  ,attention_mask=inputs['''attention_mask''']						)

						lowerCAmelCase__      :		Optional[Any]    							=      tokenizer(sentences[0]  ,return_tensors='''tf'''						).input_ids
						lowerCAmelCase__      :		Optional[Any]    							=      model.generate(input_ids=lowercase_						)

						lowerCAmelCase__      :		Optional[int]    							=      inputs_non_padded.shape[-1] - tf.math.reduce_sum(
						    tf.cast(inputs['''attention_mask'''][-1]  ,tf.intaa						)						)
						lowerCAmelCase__      :		Any    							=      tokenizer(sentences[1]  ,return_tensors='''tf'''						).input_ids
						lowerCAmelCase__      :		Optional[int]    							=      model.generate(input_ids=lowercase_  ,max_length=model.config.max_length - num_paddings						)

						lowerCAmelCase__      :		Any    							=      tokenizer.batch_decode(lowercase_  ,skip_special_tokens=lowercase_						)
						lowerCAmelCase__      :		List[Any]    							=      tokenizer.decode(output_non_padded[0]  ,skip_special_tokens=lowercase_						)
						lowerCAmelCase__      :		Optional[Any]    							=      tokenizer.decode(output_padded[0]  ,skip_special_tokens=lowercase_						)

						lowerCAmelCase__      :		List[Any]    							=      [
						    '''Hello, my dog is a little bit of a dork.\nI\'m a little bit''',
						    '''Today, I was in the middle of a conversation with a friend about the''',
						]
						self.assertListEqual(lowercase_  ,lowercase_						)
						self.assertListEqual(lowercase_  ,[non_padded_sentence, padded_sentence]						)







					def 			__lowerCAmelCase					(			self     :							List[str]						):
						lowerCAmelCase__      :		List[Any]    							=      '''facebook/opt-350m'''

						lowerCAmelCase__      :		Tuple    							=      [
						    '''Today is a beautiful day and I want to''',
						    '''In the city of San Francisco, the city''',
						    '''Paris is the capital of France and the capital''',
						    '''Computers and mobile phones have taken over the''',
						]

						lowerCAmelCase__      :		int    							=      []
						lowerCAmelCase__      :		int    							=      GPTaTokenizer.from_pretrained(lowercase_						)
						lowerCAmelCase__      :		List[str]    							=      TFOPTForCausalLM.from_pretrained(lowercase_						)

						for prompt in self.prompts:
							lowerCAmelCase__      :		Optional[Any]    							=      tokenizer(lowercase_  ,return_tensors='''tf'''						).input_ids

							lowerCAmelCase__      :		Optional[Any]    							=      model.generate(lowercase_  ,max_length=1_0						)

							lowerCAmelCase__      :		str    							=      tokenizer.batch_decode(lowercase_  ,skip_special_tokens=lowercase_						)
							predicted_outputs += generated_string

						self.assertListEqual(lowercase_  ,lowercase_						)




