
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       ):



		def __UpperCamelCase			(			self	)  -> int:
				"""simple docstring"""
				UpperCamelCase          =					'| <pad> <unk> <s> </s> a b c d e f g h i j k'.split()
				UpperCamelCase          =					dict(zip(lowerCamelCase_    ,				range(len(lowerCamelCase_	)	)	)	)

				UpperCamelCase          =					{
				    'unk_token': '<unk>',
				    'bos_token': '<s>',
				    'eos_token': '</s>',
				}
				UpperCamelCase          =					{
				    'feature_size': 1,
				    'padding_value': 0.0,
				    'sampling_rate': 16_000,
				    '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    ,				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
				UpperCamelCase          =					'hf-internal-testing/ngram-beam-search-decoder'



		def __UpperCamelCase			(			self    ,				**A_	)  -> Any:
				"""simple docstring"""
				UpperCamelCase          =					self.add_kwargs_tokens_map.copy()
				kwargs.update(lowerCamelCase_	)
				return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname    ,				**lowerCamelCase_	)



		def __UpperCamelCase			(			self    ,				**A_	)  -> Dict:
				"""simple docstring"""
				return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname    ,				**lowerCamelCase_	)



		def __UpperCamelCase			(			self    ,				**A_	)  -> List[Any]:
				"""simple docstring"""
				return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name    ,				**lowerCamelCase_	)



		def __UpperCamelCase			(			self	)  -> List[Any]:
				"""simple docstring"""
				shutil.rmtree(self.tmpdirname	)



		def __UpperCamelCase			(			self	)  -> Tuple:
				"""simple docstring"""
				UpperCamelCase          =					self.get_tokenizer()
				UpperCamelCase          =					self.get_feature_extractor()
				UpperCamelCase          =					self.get_decoder()

				UpperCamelCase          =					WavaVecaProcessorWithLM(tokenizer=lowerCamelCase_    ,				feature_extractor=lowerCamelCase_    ,				decoder=lowerCamelCase_	)

				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    ,				lowerCamelCase_	)

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

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



		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				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 __UpperCamelCase			(			self	)  -> Any:
				"""simple docstring"""
				UpperCamelCase          =					self.get_tokenizer()
				# add token to trigger raise
				tokenizer.add_tokens(['xx']	)
				with self.assertRaisesRegex(lowerCamelCase_    ,				'include'	):
						WavaVecaProcessorWithLM(
						    tokenizer=lowerCamelCase_    ,				feature_extractor=self.get_feature_extractor()    ,				decoder=self.get_decoder()	)



		def __UpperCamelCase			(			self	)  -> Optional[Any]:
				"""simple docstring"""
				UpperCamelCase          =					self.get_feature_extractor()
				UpperCamelCase          =					self.get_tokenizer()
				UpperCamelCase          =					self.get_decoder()

				UpperCamelCase          =					WavaVecaProcessorWithLM(tokenizer=lowerCamelCase_    ,				feature_extractor=lowerCamelCase_    ,				decoder=lowerCamelCase_	)

				UpperCamelCase          =					floats_list((3, 1_000)	)

				UpperCamelCase          =					feature_extractor(lowerCamelCase_    ,				return_tensors='np'	)
				UpperCamelCase          =					processor(lowerCamelCase_    ,				return_tensors='np'	)

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



		def __UpperCamelCase			(			self	)  -> Tuple:
				"""simple docstring"""
				UpperCamelCase          =					self.get_feature_extractor()
				UpperCamelCase          =					self.get_tokenizer()
				UpperCamelCase          =					self.get_decoder()

				UpperCamelCase          =					WavaVecaProcessorWithLM(tokenizer=lowerCamelCase_    ,				feature_extractor=lowerCamelCase_    ,				decoder=lowerCamelCase_	)

				UpperCamelCase          =					'This is a test string'

				UpperCamelCase          =					processor(text=lowerCamelCase_	)

				UpperCamelCase          =					tokenizer(lowerCamelCase_	)

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



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



		def __UpperCamelCase			(			self	)  -> Optional[int]:
				"""simple docstring"""
				UpperCamelCase          =					self.get_feature_extractor()
				UpperCamelCase          =					self.get_tokenizer()
				UpperCamelCase          =					self.get_decoder()

				UpperCamelCase          =					WavaVecaProcessorWithLM(tokenizer=lowerCamelCase_    ,				feature_extractor=lowerCamelCase_    ,				decoder=lowerCamelCase_	)

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

				UpperCamelCase          =					processor.decode(lowerCamelCase_	)

				UpperCamelCase          =					decoder.decode_beams(lowerCamelCase_	)[0]

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



		@parameterized.expand([[None], ['fork'], ['spawn']]	)
		def __UpperCamelCase			(			self    ,				A_	)  -> List[Any]:
				"""simple docstring"""
				UpperCamelCase          =					self.get_feature_extractor()
				UpperCamelCase          =					self.get_tokenizer()
				UpperCamelCase          =					self.get_decoder()

				UpperCamelCase          =					WavaVecaProcessorWithLM(tokenizer=lowerCamelCase_    ,				feature_extractor=lowerCamelCase_    ,				decoder=lowerCamelCase_	)

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

				UpperCamelCase          =					list(lowerCamelCase_	)

				with get_context('fork'	).Pool() as p:
						UpperCamelCase          =					decoder.decode_beams_batch(lowerCamelCase_    ,				lowerCamelCase_	)

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



		def __UpperCamelCase			(			self	)  -> Dict:
				"""simple docstring"""
				UpperCamelCase          =					self.get_feature_extractor()
				UpperCamelCase          =					self.get_tokenizer()
				UpperCamelCase          =					self.get_decoder()

				UpperCamelCase          =					WavaVecaProcessorWithLM(tokenizer=lowerCamelCase_    ,				feature_extractor=lowerCamelCase_    ,				decoder=lowerCamelCase_	)

				UpperCamelCase          =					self._get_dummy_logits()

				UpperCamelCase          =					15
				UpperCamelCase          =					-20.0
				UpperCamelCase          =					-4.0

				UpperCamelCase          =					processor.batch_decode(
				    lowerCamelCase_    ,				beam_width=lowerCamelCase_    ,				beam_prune_logp=lowerCamelCase_    ,				token_min_logp=lowerCamelCase_    ,				)
				UpperCamelCase          =					decoded_processor_out.text

				UpperCamelCase          =					list(lowerCamelCase_	)

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

				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(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.054, -18.447]    ,				lowerCamelCase_    ,				atol=1e-3	)	)

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



		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				UpperCamelCase          =					self.get_feature_extractor()
				UpperCamelCase          =					self.get_tokenizer()
				UpperCamelCase          =					self.get_decoder()

				UpperCamelCase          =					WavaVecaProcessorWithLM(tokenizer=lowerCamelCase_    ,				feature_extractor=lowerCamelCase_    ,				decoder=lowerCamelCase_	)

				UpperCamelCase          =					self._get_dummy_logits()

				UpperCamelCase          =					2.0
				UpperCamelCase          =					5.0
				UpperCamelCase          =					-20.0
				UpperCamelCase          =					True

				UpperCamelCase          =					processor.batch_decode(
				    lowerCamelCase_    ,				alpha=lowerCamelCase_    ,				beta=lowerCamelCase_    ,				unk_score_offset=lowerCamelCase_    ,				lm_score_boundary=lowerCamelCase_    ,				)
				UpperCamelCase          =					decoded_processor_out.text

				UpperCamelCase          =					list(lowerCamelCase_	)
				decoder.reset_params(
				    alpha=lowerCamelCase_    ,				beta=lowerCamelCase_    ,				unk_score_offset=lowerCamelCase_    ,				lm_score_boundary=lowerCamelCase_    ,				)

				with get_context('fork'	).Pool() as pool:
						UpperCamelCase          =					decoder.decode_beams_batch(
						    lowerCamelCase_    ,				lowerCamelCase_    ,				)

				UpperCamelCase          =					[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_	)
				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    ,				lowerCamelCase_	)



		def __UpperCamelCase			(			self	)  -> Optional[int]:
				"""simple docstring"""
				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(lowerCamelCase_	)
				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(lowerCamelCase_    ,				lowerCamelCase_	)



		def __UpperCamelCase			(			self	)  -> Optional[int]:
				"""simple docstring"""
				UpperCamelCase          =					snapshot_download('hf-internal-testing/processor_with_lm'	)

				UpperCamelCase          =					WavaVecaProcessorWithLM.from_pretrained(lowerCamelCase_	)

				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(lowerCamelCase_	)
				UpperCamelCase          =					os.listdir(lowerCamelCase_	)

				local_decoder_files.sort()
				expected_decoder_files.sort()

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



		def __UpperCamelCase			(			self	)  -> str:
				"""simple docstring"""
				UpperCamelCase          =					WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'	)
				UpperCamelCase          =					AutoProcessor.from_pretrained('hf-internal-testing/processor_with_lm'	)

				UpperCamelCase          =					floats_list((3, 1_000)	)

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

				UpperCamelCase          =					self._get_dummy_logits()

				UpperCamelCase          =					processor_wavaveca.batch_decode(lowerCamelCase_	)
				UpperCamelCase          =					processor_auto.batch_decode(lowerCamelCase_	)

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



		def __UpperCamelCase			(			self	)  -> List[Any]:
				"""simple docstring"""
				UpperCamelCase          =					self.get_feature_extractor()
				UpperCamelCase          =					self.get_tokenizer()
				UpperCamelCase          =					self.get_decoder()

				UpperCamelCase          =					WavaVecaProcessorWithLM(tokenizer=lowerCamelCase_    ,				feature_extractor=lowerCamelCase_    ,				decoder=lowerCamelCase_	)

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



		@staticmethod
		def __UpperCamelCase			(			A_    ,				A_	)  -> Dict:
				"""simple docstring"""
				UpperCamelCase          =					[d[key] for d in offsets]
				return retrieved_list



		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				UpperCamelCase          =					WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'	)
				UpperCamelCase          =					self._get_dummy_logits()[0]

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

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



		def __UpperCamelCase			(			self	)  -> Optional[Any]:
				"""simple docstring"""
				UpperCamelCase          =					WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'	)
				UpperCamelCase          =					self._get_dummy_logits()

				UpperCamelCase          =					processor.batch_decode(lowerCamelCase_    ,				output_word_offsets=lowerCamelCase_	)

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

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



		@slow
		@require_torch
		@require_torchaudio
		def __UpperCamelCase			(			self	)  -> Optional[int]:
				"""simple docstring"""
				import torch

				UpperCamelCase          =					load_dataset('common_voice'    ,				'en'    ,				split='train'    ,				streaming=lowerCamelCase_	)
				UpperCamelCase          =					ds.cast_column('audio'    ,				datasets.Audio(sampling_rate=16_000	)	)
				UpperCamelCase          =					iter(lowerCamelCase_	)
				UpperCamelCase          =					next(lowerCamelCase_	)

				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(lowerCamelCase_	).logits.cpu().numpy()

				UpperCamelCase          =					processor.decode(logits[0]    ,				output_word_offsets=lowerCamelCase_	)

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

				# output times
				UpperCamelCase          =					torch.tensor(self.get_from_offsets(lowerCamelCase_    ,				'start_time'	)	)
				UpperCamelCase          =					torch.tensor(self.get_from_offsets(lowerCamelCase_    ,				'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(lowerCamelCase_    ,				lowerCamelCase_    ,				atol=0.01	)	)
				self.assertTrue(torch.allclose(lowerCamelCase_    ,				lowerCamelCase_    ,				atol=0.01	)	)


import unittest

from parameterized import parameterized

from transformers import LlamaConfig, is_torch_available, set_seed
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
					import torch

					from transformers import LlamaForCausalLM, LlamaForSequenceClassification, LlamaModel, LlamaTokenizer


class     lowercase :



		def __init__(			self    ,				A_    ,				A_=13    ,				A_=7    ,				A_=True    ,				A_=True    ,				A_=False    ,				A_=True    ,				A_=99    ,				A_=32    ,				A_=5    ,				A_=4    ,				A_=37    ,				A_="gelu"    ,				A_=0.1    ,				A_=0.1    ,				A_=512    ,				A_=16    ,				A_=2    ,				A_=0.02    ,				A_=3    ,				A_=4    ,				A_=None    ,				)  -> Tuple:
				"""simple docstring"""
				UpperCamelCase          =					parent
				UpperCamelCase          =					batch_size
				UpperCamelCase          =					seq_length
				UpperCamelCase          =					is_training
				UpperCamelCase          =					use_input_mask
				UpperCamelCase          =					use_token_type_ids
				UpperCamelCase          =					use_labels
				UpperCamelCase          =					vocab_size
				UpperCamelCase          =					hidden_size
				UpperCamelCase          =					num_hidden_layers
				UpperCamelCase          =					num_attention_heads
				UpperCamelCase          =					intermediate_size
				UpperCamelCase          =					hidden_act
				UpperCamelCase          =					hidden_dropout_prob
				UpperCamelCase          =					attention_probs_dropout_prob
				UpperCamelCase          =					max_position_embeddings
				UpperCamelCase          =					type_vocab_size
				UpperCamelCase          =					type_sequence_label_size
				UpperCamelCase          =					initializer_range
				UpperCamelCase          =					num_labels
				UpperCamelCase          =					num_choices
				UpperCamelCase          =					scope



		def __UpperCamelCase			(			self	)  -> Any:
				"""simple docstring"""
				UpperCamelCase          =					ids_tensor([self.batch_size, self.seq_length]    ,				self.vocab_size	)

				UpperCamelCase          =					None
				if self.use_input_mask:
						UpperCamelCase          =					random_attention_mask([self.batch_size, self.seq_length]	)

				UpperCamelCase          =					None
				if self.use_token_type_ids:
						UpperCamelCase          =					ids_tensor([self.batch_size, self.seq_length]    ,				self.type_vocab_size	)

				UpperCamelCase          =					None
				UpperCamelCase          =					None
				UpperCamelCase          =					None
				if self.use_labels:
						UpperCamelCase          =					ids_tensor([self.batch_size]    ,				self.type_sequence_label_size	)
						UpperCamelCase          =					ids_tensor([self.batch_size, self.seq_length]    ,				self.num_labels	)
						UpperCamelCase          =					ids_tensor([self.batch_size]    ,				self.num_choices	)

				UpperCamelCase          =					self.get_config()

				return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels



		def __UpperCamelCase			(			self	)  -> List[Any]:
				"""simple docstring"""
				return LlamaConfig(
				    vocab_size=self.vocab_size    ,				hidden_size=self.hidden_size    ,				num_hidden_layers=self.num_hidden_layers    ,				num_attention_heads=self.num_attention_heads    ,				intermediate_size=self.intermediate_size    ,				hidden_act=self.hidden_act    ,				hidden_dropout_prob=self.hidden_dropout_prob    ,				attention_probs_dropout_prob=self.attention_probs_dropout_prob    ,				max_position_embeddings=self.max_position_embeddings    ,				type_vocab_size=self.type_vocab_size    ,				is_decoder=A_    ,				initializer_range=self.initializer_range    ,				)



		def __UpperCamelCase			(			self    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_	)  -> int:
				"""simple docstring"""
				UpperCamelCase          =					LlamaModel(config=A_	)
				model.to(A_	)
				model.eval()
				UpperCamelCase          =					model(A_    ,				attention_mask=A_	)
				UpperCamelCase          =					model(A_	)
				self.parent.assertEqual(result.last_hidden_state.shape    ,				(self.batch_size, self.seq_length, self.hidden_size)	)



		def __UpperCamelCase			(			self    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				)  -> List[Any]:
				"""simple docstring"""
				UpperCamelCase          =					True
				UpperCamelCase          =					LlamaModel(A_	)
				model.to(A_	)
				model.eval()
				UpperCamelCase          =					model(
				    A_    ,				attention_mask=A_    ,				encoder_hidden_states=A_    ,				encoder_attention_mask=A_    ,				)
				UpperCamelCase          =					model(
				    A_    ,				attention_mask=A_    ,				encoder_hidden_states=A_    ,				)
				UpperCamelCase          =					model(A_    ,				attention_mask=A_	)
				self.parent.assertEqual(result.last_hidden_state.shape    ,				(self.batch_size, self.seq_length, self.hidden_size)	)



		def __UpperCamelCase			(			self    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				)  -> str:
				"""simple docstring"""
				UpperCamelCase          =					LlamaForCausalLM(config=A_	)
				model.to(A_	)
				model.eval()
				UpperCamelCase          =					model(A_    ,				attention_mask=A_    ,				labels=A_	)
				self.parent.assertEqual(result.logits.shape    ,				(self.batch_size, self.seq_length, self.vocab_size)	)



		def __UpperCamelCase			(			self    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				A_    ,				)  -> Optional[int]:
				"""simple docstring"""
				UpperCamelCase          =					True
				UpperCamelCase          =					True
				UpperCamelCase          =					LlamaForCausalLM(config=A_	)
				model.to(A_	)
				model.eval()

				# first forward pass
				UpperCamelCase          =					model(
				    A_    ,				attention_mask=A_    ,				encoder_hidden_states=A_    ,				encoder_attention_mask=A_    ,				use_cache=A_    ,				)
				UpperCamelCase          =					outputs.past_key_values

				# create hypothetical multiple next token and extent to next_input_ids
				UpperCamelCase          =					ids_tensor((self.batch_size, 3)    ,				config.vocab_size	)
				UpperCamelCase          =					ids_tensor((self.batch_size, 3)    ,				vocab_size=2	)

				# append to next input_ids and
				UpperCamelCase          =					torch.cat([input_ids, next_tokens]    ,				dim=-1	)
				UpperCamelCase          =					torch.cat([input_mask, next_mask]    ,				dim=-1	)

				UpperCamelCase          =					model(
				    A_    ,				attention_mask=A_    ,				encoder_hidden_states=A_    ,				encoder_attention_mask=A_    ,				output_hidden_states=A_    ,				)['hidden_states'][0]
				UpperCamelCase          =					model(
				    A_    ,				attention_mask=A_    ,				encoder_hidden_states=A_    ,				encoder_attention_mask=A_    ,				past_key_values=A_    ,				output_hidden_states=A_    ,				)['hidden_states'][0]

				# select random slice
				UpperCamelCase          =					ids_tensor((1,)    ,				output_from_past.shape[-1]	).item()
				UpperCamelCase          =					output_from_no_past[:, -3:, random_slice_idx].detach()
				UpperCamelCase          =					output_from_past[:, :, random_slice_idx].detach()

				self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]	)

				# test that outputs are equal for slice
				self.parent.assertTrue(torch.allclose(A_    ,				A_    ,				atol=1e-3	)	)



		def __UpperCamelCase			(			self	)  -> Tuple:
				"""simple docstring"""
				UpperCamelCase          =					self.prepare_config_and_inputs()
				(
				    (
				    UpperCamelCase
				)     ,  (
				    UpperCamelCase
				)     ,  (
				    UpperCamelCase
				)     ,  (
				    UpperCamelCase
				)     ,  (
				    UpperCamelCase
				)     ,  (
				    UpperCamelCase
				)     ,  (
				    UpperCamelCase
				)     ,  
				)          =					config_and_inputs
				UpperCamelCase          =					{'input_ids': input_ids, 'attention_mask': input_mask}
				return config, inputs_dict






@require_torch
class     lowercase (		_SCREAMING_SNAKE_CASE  ,       _SCREAMING_SNAKE_CASE  ,       _SCREAMING_SNAKE_CASE  ,       unittest.TestCase       ):
		__lowercase	:	str     					=    (LlamaModel, LlamaForCausalLM, LlamaForSequenceClassification) if is_torch_available() else ()
		__lowercase	:	str     					=    (LlamaForCausalLM,) if is_torch_available() else ()
		__lowercase	:	Any     					=    (
		    {
		        "feature-extraction": LlamaModel,
		        "text-classification": LlamaForSequenceClassification,
		        "text-generation": LlamaForCausalLM,
		        "zero-shot": LlamaForSequenceClassification,
		    }
		    if is_torch_available()
		    else {}
		)
		__lowercase	:	int     					=    False
		__lowercase	:	Optional[int]     					=    False



		def __UpperCamelCase			(			self	)  -> int:
				"""simple docstring"""
				UpperCamelCase          =					LlamaModelTester(self	)
				UpperCamelCase          =					ConfigTester(self    ,				config_class=A_    ,				hidden_size=37	)



		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				self.config_tester.run_common_tests()



		def __UpperCamelCase			(			self	)  -> Optional[int]:
				"""simple docstring"""
				UpperCamelCase          =					self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_model(*A_	)



		def __UpperCamelCase			(			self	)  -> Optional[Any]:
				"""simple docstring"""
				UpperCamelCase          =					self.model_tester.prepare_config_and_inputs()
				for type in ["absolute", "relative_key", "relative_key_query"]:
						UpperCamelCase          =					type
						self.model_tester.create_and_check_model(*A_	)



		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				UpperCamelCase     ,  UpperCamelCase          =					self.model_tester.prepare_config_and_inputs_for_common()
				UpperCamelCase          =					3
				UpperCamelCase          =					input_dict['input_ids']
				UpperCamelCase          =					input_ids.ne(1	).to(A_	)
				UpperCamelCase          =					ids_tensor([self.model_tester.batch_size]    ,				self.model_tester.type_sequence_label_size	)
				UpperCamelCase          =					LlamaForSequenceClassification(A_	)
				model.to(A_	)
				model.eval()
				UpperCamelCase          =					model(A_    ,				attention_mask=A_    ,				labels=A_	)
				self.assertEqual(result.logits.shape    ,				(self.model_tester.batch_size, self.model_tester.num_labels)	)



		def __UpperCamelCase			(			self	)  -> Dict:
				"""simple docstring"""
				UpperCamelCase     ,  UpperCamelCase          =					self.model_tester.prepare_config_and_inputs_for_common()
				UpperCamelCase          =					3
				UpperCamelCase          =					'single_label_classification'
				UpperCamelCase          =					input_dict['input_ids']
				UpperCamelCase          =					input_ids.ne(1	).to(A_	)
				UpperCamelCase          =					ids_tensor([self.model_tester.batch_size]    ,				self.model_tester.type_sequence_label_size	)
				UpperCamelCase          =					LlamaForSequenceClassification(A_	)
				model.to(A_	)
				model.eval()
				UpperCamelCase          =					model(A_    ,				attention_mask=A_    ,				labels=A_	)
				self.assertEqual(result.logits.shape    ,				(self.model_tester.batch_size, self.model_tester.num_labels)	)



		def __UpperCamelCase			(			self	)  -> List[str]:
				"""simple docstring"""
				UpperCamelCase     ,  UpperCamelCase          =					self.model_tester.prepare_config_and_inputs_for_common()
				UpperCamelCase          =					3
				UpperCamelCase          =					'multi_label_classification'
				UpperCamelCase          =					input_dict['input_ids']
				UpperCamelCase          =					input_ids.ne(1	).to(A_	)
				UpperCamelCase          =					ids_tensor(
				    [self.model_tester.batch_size, config.num_labels]    ,				self.model_tester.type_sequence_label_size	).to(torch.float	)
				UpperCamelCase          =					LlamaForSequenceClassification(A_	)
				model.to(A_	)
				model.eval()
				UpperCamelCase          =					model(A_    ,				attention_mask=A_    ,				labels=A_	)
				self.assertEqual(result.logits.shape    ,				(self.model_tester.batch_size, self.model_tester.num_labels)	)



		@unittest.skip('LLaMA buffers include complex numbers, which breaks this test'	)
		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				pass



		@parameterized.expand([('linear',), ('dynamic',)]	)
		def __UpperCamelCase			(			self    ,				A_	)  -> Optional[int]:
				"""simple docstring"""
				UpperCamelCase     ,  UpperCamelCase          =					self.model_tester.prepare_config_and_inputs_for_common()
				UpperCamelCase          =					ids_tensor([1, 10]    ,				config.vocab_size	)
				UpperCamelCase          =					ids_tensor([1, int(config.max_position_embeddings * 1.5	)]    ,				config.vocab_size	)

				set_seed(42	)  # Fixed seed at init time so the two models get the same random weights
				UpperCamelCase          =					LlamaModel(A_	)
				original_model.to(A_	)
				original_model.eval()
				UpperCamelCase          =					original_model(A_	).last_hidden_state
				UpperCamelCase          =					original_model(A_	).last_hidden_state

				set_seed(42	)  # Fixed seed at init time so the two models get the same random weights
				UpperCamelCase          =					{'type': scaling_type, 'factor': 10.0}
				UpperCamelCase          =					LlamaModel(A_	)
				scaled_model.to(A_	)
				scaled_model.eval()
				UpperCamelCase          =					scaled_model(A_	).last_hidden_state
				UpperCamelCase          =					scaled_model(A_	).last_hidden_state

				# Dynamic scaling does not change the RoPE embeddings until it receives an input longer than the original
				# maximum sequence length, so the outputs for the short input should match.
				if scaling_type == "dynamic":
						self.assertTrue(torch.allclose(A_    ,				A_    ,				atol=1e-5	)	)
				else:
						self.assertFalse(torch.allclose(A_    ,				A_    ,				atol=1e-5	)	)

				# The output should be different for long inputs
				self.assertFalse(torch.allclose(A_    ,				A_    ,				atol=1e-5	)	)






@require_torch
class     lowercase (		unittest.TestCase       ):



		@unittest.skip('Logits are not exactly the same, once we fix the instabalities somehow, will update!'	)
		@slow
		def __UpperCamelCase			(			self	)  -> List[Any]:
				"""simple docstring"""
				UpperCamelCase          =					[1, 306, 4_658, 278, 6_593, 310, 2_834, 338]
				UpperCamelCase          =					LlamaForCausalLM.from_pretrained('meta-llama/Llama-2-7b-hf'    ,				device_map='auto'	)
				UpperCamelCase          =					model(torch.tensor([input_ids]	)	)
				# Expected mean on dim = -1
				UpperCamelCase          =					torch.tensor([[-6.6550, -4.1227, -4.9859, -3.2406, 0.8262, -3.0033, 1.2964, -3.3699]]	)
				torch.testing.assert_close(out.mean(-1	)    ,				A_    ,				atol=1e-2    ,				rtol=1e-2	)
				# slicing logits[0, 0, 0:30]
				# fmt: off
				UpperCamelCase          =					torch.tensor([-12.8281, -7.4453, -0.4639, -8.0625, -7.2500, -8.0000, -6.4883, -7.7695, -7.8438, -7.0312, -6.2188, -7.1328, -1.8496, 1.9961, -8.6250, -6.7227, -12.8281, -6.9492, -7.0742, -7.7852, -7.5820, -7.9062, -6.9375, -7.9805, -8.3438, -8.1562, -8.0469, -7.6250, -7.7422, -7.3398,]	)
				# fmt: on
				torch.testing.assert_close(out[0, 0, :30]    ,				A_    ,				atol=1e-5    ,				rtol=1e-5	)



		@unittest.skip('Logits are not exactly the same, once we fix the instabalities somehow, will update!'	)
		@slow
		def __UpperCamelCase			(			self	)  -> List[Any]:
				"""simple docstring"""
				UpperCamelCase          =					[1, 306, 4_658, 278, 6_593, 310, 2_834, 338]
				UpperCamelCase          =					LlamaForCausalLM.from_pretrained('meta-llama/Llama-2-13b-hf'    ,				device_map='auto'	)
				UpperCamelCase          =					model(torch.tensor(A_	)	)
				# Expected mean on dim = -1
				UpperCamelCase          =					torch.tensor([[-2.0622, -1.2794, -1.1638, -0.9788, -1.4603, -1.0238, -1.7893, -1.4411]]	)
				torch.testing.assert_close(out.mean(-1	)    ,				A_    ,				atol=1e-2    ,				rtol=1e-2	)
				# slicing logits[0, 0, 0:30]
				# fmt: off
				UpperCamelCase          =					torch.tensor([-8.1406, -8.0547, 2.7461, -1.2344, -0.1448, -1.8262, -1.0020, -1.8154, -1.6895, -1.8516, -2.3574, -0.9277, 3.7598, 6.5742, -1.2998, -0.1177, -8.1406, -2.9688, -2.9199, -3.1699, -3.5254, -2.3555, -2.7988, -3.4141, -2.8262, -4.5195, -3.3379, -3.3164, -2.7832, -3.0273]	)
				# fmt: on
				torch.testing.assert_close(out[0, 0, :30]    ,				A_    ,				atol=1e-5    ,				rtol=1e-5	)



		@unittest.skip('Logits are not exactly the same, once we fix the instabalities somehow, will update!'	)
		@slow
		def __UpperCamelCase			(			self	)  -> List[Any]:
				"""simple docstring"""
				UpperCamelCase          =					[1, 306, 4_658, 278, 6_593, 310, 2_834, 338]
				UpperCamelCase          =					LlamaForCausalLM.from_pretrained('meta-llama/Llama-2-13b-chat-hf'    ,				device_map='auto'	)
				UpperCamelCase          =					model(torch.tensor(A_	)	)
				# Expected mean on dim = -1
				UpperCamelCase          =					torch.tensor([[-0.8562, -1.8520, -0.7551, -0.4162, -1.5161, -1.2038, -2.4823, -2.3254]]	)
				torch.testing.assert_close(out.mean(-1	)    ,				A_    ,				atol=1e-2    ,				rtol=1e-2	)
				# slicing logits[0, 0, 0:30]
				# fmt: off
				UpperCamelCase          =					torch.tensor([-2.2227, 4.8828, 0.9023, -0.4578, -0.7871, -0.1033, -0.6221, -0.5786, -0.7803, -1.0674, -1.2920, -0.1570, 0.8008, 2.0723, -0.9497, 0.2771, -2.2227, -0.7612, -1.4346, -1.2061, -1.6426, -0.3000, -0.7139, -1.1934, -1.8691, -1.6973, -1.5947, -1.2705, -0.3523, -0.5513]	)
				# fmt: on
				torch.testing.assert_close(out.mean(-1	)    ,				A_    ,				atol=1e-2    ,				rtol=1e-2	)



		@unittest.skip(
		    'Logits are not exactly the same, once we fix the instabalities somehow, will update! Also it is gonna be a `too_slow` test'	)
		@slow
		def __UpperCamelCase			(			self	)  -> List[str]:
				"""simple docstring"""
				UpperCamelCase          =					[1, 306, 4_658, 278, 6_593, 310, 2_834, 338]
				UpperCamelCase          =					LlamaForCausalLM.from_pretrained('meta-llama/Llama-2-70b-hf'    ,				device_map='auto'	)
				UpperCamelCase          =					model(torch.tensor(A_	)	)

				UpperCamelCase          =					torch.tensor(
				    [[-4.2327, -3.3360, -4.6665, -4.7631, -1.8180, -3.4170, -1.4211, -3.1810]]    ,				dtype=torch.floataa	)
				torch.testing.assert_close(out.mean(-1	)    ,				A_    ,				atol=1e-2    ,				rtol=1e-2	)
				# fmt: off
				UpperCamelCase          =					torch.tensor([-9.4922, -3.9551, 1.7998, -5.6758, -5.1055, -5.8984, -4.8320, -6.8086, -6.5391, -5.6172, -5.5820, -5.5352, 1.7881, 3.6289, -6.5117, -3.4785, -9.5000, -6.0352, -6.8125, -6.0195, -6.6836, -5.4727, -6.2812, -6.0391, -7.3398, -7.4297, -7.4844, -6.5820, -5.8789, -5.5312]	)
				# fmt: on
				torch.testing.assert_close(out[0, 0, :30]    ,				A_    ,				atol=1e-5    ,				rtol=1e-5	)



		@unittest.skip('Model is curently gated'	)
		@slow
		def __UpperCamelCase			(			self	)  -> Tuple:
				"""simple docstring"""
				UpperCamelCase          =					'Simply put, the theory of relativity states that 1) the laws of physics are the same everywhere in the universe and 2) the passage of time and the length of objects can vary depending on the observer\'s frame of reference.\n\nThe first part of the theory, that the laws of physics are the same everywhere, is known as the "princi'
				UpperCamelCase          =					'Simply put, the theory of relativity states that '
				UpperCamelCase          =					LlamaTokenizer.from_pretrained('meta-llama/Llama-2-13b-chat-hf'	)
				UpperCamelCase          =					tokenizer.encode(A_    ,				return_tensors='pt'	)
				UpperCamelCase          =					LlamaForCausalLM.from_pretrained(
				    'meta-llama/Llama-2-13b-chat-hf'    ,				device_map='sequential'    ,				use_safetensors=A_	)

				# greedy generation outputs
				UpperCamelCase          =					model.generate(A_    ,				max_new_tokens=64    ,				top_p=A_    ,				temperature=1    ,				do_sample=A_	)
				UpperCamelCase          =					tokenizer.decode(generated_ids[0]    ,				skip_special_tokens=A_	)
				self.assertEqual(A_    ,				A_	)
