






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





				"""simple docstring"""







				def       __lowercase							( self    ):




									"""simple docstring"""

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

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

									__UpperCamelCase      :					Optional[Any]       		=tempfile.mkdtemp()
									__UpperCamelCase      :					Optional[int]       		=os.path.join(self.tmpdirname							,  VOCAB_FILES_NAMES['vocab_file']    )
									__UpperCamelCase      :					List[Any]       		=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      :					Union[str, Any]       		='hf-internal-testing/ngram-beam-search-decoder'


				def       __lowercase							( self							,  **lowerCamelCase__    ):




									"""simple docstring"""

									__UpperCamelCase      :					Any       		=self.add_kwargs_tokens_map.copy()
									kwargs.update(lowerCAmelCase_    )
									return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname							,  **lowerCAmelCase_    )


				def       __lowercase							( self							,  **lowerCamelCase__    ):




									"""simple docstring"""

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


				def       __lowercase							( self							,  **lowerCamelCase__    ):




									"""simple docstring"""

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


				def       __lowercase							( self    ):




									"""simple docstring"""

									shutil.rmtree(self.tmpdirname    )


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Any       		=self.get_tokenizer()
									__UpperCamelCase      :					Optional[int]       		=self.get_feature_extractor()
									__UpperCamelCase      :					List[Any]       		=self.get_decoder()

									__UpperCamelCase      :					Optional[int]       		=WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_							,  feature_extractor=lowerCAmelCase_							,  decoder=lowerCAmelCase_    )

									processor.save_pretrained(self.tmpdirname    )
									__UpperCamelCase      :					Dict       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Dict       		=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      :					Tuple       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Any       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Optional[int]       		=self.get_feature_extractor()
									__UpperCamelCase      :					Tuple       		=self.get_tokenizer()
									__UpperCamelCase      :					List[Any]       		=self.get_decoder()

									__UpperCamelCase      :					Optional[Any]       		=WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_							,  feature_extractor=lowerCAmelCase_							,  decoder=lowerCAmelCase_    )

									__UpperCamelCase      :					Union[str, Any]       		=floats_list((3, 1000)    )

									__UpperCamelCase      :					str       		=feature_extractor(lowerCAmelCase_							,  return_tensors='np'    )
									__UpperCamelCase      :					Dict       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Tuple       		=self.get_feature_extractor()
									__UpperCamelCase      :					int       		=self.get_tokenizer()
									__UpperCamelCase      :					List[str]       		=self.get_decoder()

									__UpperCamelCase      :					Optional[Any]       		=WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_							,  feature_extractor=lowerCAmelCase_							,  decoder=lowerCAmelCase_    )

									__UpperCamelCase      :					Optional[int]       		='This is a test string'

									__UpperCamelCase      :					str       		=processor(text=lowerCAmelCase_    )

									__UpperCamelCase      :					Optional[int]       		=tokenizer(lowerCAmelCase_    )

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


				def       __lowercase							( self							,  lowerCamelCase__=(2, 10, 16)							,  lowerCamelCase__=77    ):




									"""simple docstring"""

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


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					str       		=self.get_feature_extractor()
									__UpperCamelCase      :					Optional[int]       		=self.get_tokenizer()
									__UpperCamelCase      :					List[Any]       		=self.get_decoder()

									__UpperCamelCase      :					Tuple       		=WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_							,  feature_extractor=lowerCAmelCase_							,  decoder=lowerCAmelCase_    )

									__UpperCamelCase      :					str       		=self._get_dummy_logits(shape=(10, 16)							,  seed=13    )

									__UpperCamelCase      :					int       		=processor.decode(lowerCAmelCase_    )

									__UpperCamelCase      :					int       		=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       __lowercase							( self							,  lowerCamelCase__    ):




									"""simple docstring"""

									__UpperCamelCase      :					Optional[int]       		=self.get_feature_extractor()
									__UpperCamelCase      :					int       		=self.get_tokenizer()
									__UpperCamelCase      :					List[Any]       		=self.get_decoder()

									__UpperCamelCase      :					List[Any]       		=WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_							,  feature_extractor=lowerCAmelCase_							,  decoder=lowerCAmelCase_    )

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

									__UpperCamelCase      :					List[Any]       		=list(lowerCAmelCase_    )

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

									__UpperCamelCase		,				__UpperCamelCase		,				__UpperCamelCase      :					int       		=[], [], []
									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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Optional[int]       		=self.get_feature_extractor()
									__UpperCamelCase      :					Optional[int]       		=self.get_tokenizer()
									__UpperCamelCase      :					Tuple       		=self.get_decoder()

									__UpperCamelCase      :					Optional[Any]       		=WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_							,  feature_extractor=lowerCAmelCase_							,  decoder=lowerCAmelCase_    )

									__UpperCamelCase      :					Any       		=self._get_dummy_logits()

									__UpperCamelCase      :					Optional[Any]       		=15
									__UpperCamelCase      :					Tuple       		=-20.0
									__UpperCamelCase      :					Dict       		=-4.0

									__UpperCamelCase      :					Tuple       		=processor.batch_decode(
									    lowerCAmelCase_							,  beam_width=lowerCAmelCase_							,  beam_prune_logp=lowerCAmelCase_							,  token_min_logp=lowerCAmelCase_							,  )
									__UpperCamelCase      :					Dict       		=decoded_processor_out.text

									__UpperCamelCase      :					List[str]       		=list(lowerCAmelCase_    )

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

									__UpperCamelCase      :					int       		=[d[0][0] for d in decoded_decoder_out]
									__UpperCamelCase      :					List[Any]       		=[d[0][2] for d in decoded_decoder_out]
									__UpperCamelCase      :					Dict       		=[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.9_474]							,  lowerCAmelCase_							,  atol=1E-3    )    )


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					int       		=self.get_feature_extractor()
									__UpperCamelCase      :					Dict       		=self.get_tokenizer()
									__UpperCamelCase      :					int       		=self.get_decoder()

									__UpperCamelCase      :					List[Any]       		=WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_							,  feature_extractor=lowerCAmelCase_							,  decoder=lowerCAmelCase_    )

									__UpperCamelCase      :					Optional[Any]       		=self._get_dummy_logits()

									__UpperCamelCase      :					Any       		=2.0
									__UpperCamelCase      :					List[Any]       		=5.0
									__UpperCamelCase      :					str       		=-20.0
									__UpperCamelCase      :					str       		=True

									__UpperCamelCase      :					str       		=processor.batch_decode(
									    lowerCAmelCase_							,  alpha=lowerCAmelCase_							,  beta=lowerCAmelCase_							,  unk_score_offset=lowerCAmelCase_							,  lm_score_boundary=lowerCAmelCase_							,  )
									__UpperCamelCase      :					List[str]       		=decoded_processor_out.text

									__UpperCamelCase      :					Dict       		=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      :					str       		=decoder.decode_beams_batch(
														    lowerCAmelCase_							,  lowerCAmelCase_							,  )

									__UpperCamelCase      :					Tuple       		=[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      :					Tuple       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					List[str]       		=WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'    )

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

									__UpperCamelCase      :					Optional[int]       		=os.listdir(lowerCAmelCase_    )
									__UpperCamelCase      :					Tuple       		=['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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					List[Any]       		=snapshot_download('hf-internal-testing/processor_with_lm'    )

									__UpperCamelCase      :					List[str]       		=WavaVecaProcessorWithLM.from_pretrained(lowerCAmelCase_    )

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

									__UpperCamelCase      :					Tuple       		=os.listdir(lowerCAmelCase_    )
									__UpperCamelCase      :					str       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Tuple       		=WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'    )
									__UpperCamelCase      :					Optional[Any]       		=AutoProcessor.from_pretrained('hf-internal-testing/processor_with_lm'    )

									__UpperCamelCase      :					Optional[int]       		=floats_list((3, 1000)    )

									__UpperCamelCase      :					Union[str, Any]       		=processor_wavaveca(lowerCAmelCase_							,  return_tensors='np'    )
									__UpperCamelCase      :					Dict       		=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      :					Any       		=self._get_dummy_logits()

									__UpperCamelCase      :					int       		=processor_wavaveca.batch_decode(lowerCAmelCase_    )
									__UpperCamelCase      :					Union[str, Any]       		=processor_auto.batch_decode(lowerCAmelCase_    )

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


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					List[Any]       		=self.get_feature_extractor()
									__UpperCamelCase      :					Dict       		=self.get_tokenizer()
									__UpperCamelCase      :					Dict       		=self.get_decoder()

									__UpperCamelCase      :					Any       		=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       __lowercase							( lowerCamelCase__							,  lowerCamelCase__    ):




									"""simple docstring"""

									__UpperCamelCase      :					Dict       		=[d[key] for d in offsets]
									return retrieved_list


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					str       		=WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'    )
									__UpperCamelCase      :					Any       		=self._get_dummy_logits()[0]

									__UpperCamelCase      :					List[str]       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					str       		=WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'    )
									__UpperCamelCase      :					List[str]       		=self._get_dummy_logits()

									__UpperCamelCase      :					Dict       		=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       __lowercase							( self    ):




									"""simple docstring"""

									import torch

									__UpperCamelCase      :					Any       		=load_dataset('common_voice'							,  'en'							,  split='train'							,  streaming=lowerCAmelCase_    )
									__UpperCamelCase      :					Tuple       		=ds.cast_column('audio'							,  datasets.Audio(sampling_rate=16000    )    )
									__UpperCamelCase      :					int       		=iter(lowerCAmelCase_    )
									__UpperCamelCase      :					Any       		=next(lowerCAmelCase_    )

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

									with torch.no_grad():
														__UpperCamelCase      :					Optional[Any]       		=model(lowerCAmelCase_    ).logits.cpu().numpy()

									__UpperCamelCase      :					Dict       		=processor.decode(logits[0]							,  output_word_offsets=lowerCAmelCase_    )

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

									__UpperCamelCase      :					Any       		='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      :					Any       		=torch.tensor(self.get_from_offsets(lowerCAmelCase_							,  'start_time'    )    )
									__UpperCamelCase      :					Any       		=torch.tensor(self.get_from_offsets(lowerCAmelCase_							,  'end_time'    )    )

									# fmt: off
									__UpperCamelCase      :					Optional[int]       		=torch.tensor([1.4_199, 1.6_599, 2.2_599, 3.0, 3.24, 3.5_999, 3.7_999, 4.0_999, 4.26, 4.94, 5.28, 5.6_599, 5.78, 5.94, 6.32, 6.5_399, 6.6_599]    )
									__UpperCamelCase      :					List[Any]       		=torch.tensor([1.5_399, 1.8_999, 2.9, 3.16, 3.5_399, 3.72, 4.0_199, 4.1_799, 4.76, 5.1_599, 5.5_599, 5.6_999, 5.86, 6.1_999, 6.38, 6.6_199, 6.94]    )
									# fmt: on

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


import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers.testing_utils import require_vision
from transformers.utils import is_vision_available


if is_vision_available():
    from PIL import Image

    from transformers import (
        AutoProcessor,
        BertTokenizerFast,
        BlipImageProcessor,
        GPTaTokenizer,
        InstructBlipProcessor,
        PreTrainedTokenizerFast,
    )



@require_vision
class 				a							(			unittest.TestCase       ):







 """simple docstring"""






 def      UpperCAmelCase							(					self							)  ->					List[Any]:
        _A        =  tempfile.mkdtemp()

        _A        =  BlipImageProcessor()
        _A        =  GPTaTokenizer.from_pretrained("""hf-internal-testing/tiny-random-GPT2Model"""							)
        _A        =  BertTokenizerFast.from_pretrained("""hf-internal-testing/tiny-random-bert"""							)

        _A        =  InstructBlipProcessor(lowerCAmelCase_		,				lowerCAmelCase_		,				lowerCAmelCase_							)

        processor.save_pretrained(self.tmpdirname							)



 def      UpperCAmelCase							(					self		,				**lowerCAmelCase_							)  ->					Any:
        return AutoProcessor.from_pretrained(self.tmpdirname		,				**lowerCAmelCase_							).tokenizer



 def      UpperCAmelCase							(					self		,				**lowerCAmelCase_							)  ->					Dict:
        return AutoProcessor.from_pretrained(self.tmpdirname		,				**lowerCAmelCase_							).image_processor



 def      UpperCAmelCase							(					self		,				**lowerCAmelCase_							)  ->					Union[str, Any]:
        return AutoProcessor.from_pretrained(self.tmpdirname		,				**lowerCAmelCase_							).qformer_tokenizer



 def      UpperCAmelCase							(					self							)  ->					int:
        shutil.rmtree(self.tmpdirname							)



 def      UpperCAmelCase							(					self							)  ->					Union[str, Any]:
        _A        =  [np.random.randint(2_55		,				size=(3, 30, 4_00)		,				dtype=np.uinta							)]

        _A        =  [Image.fromarray(np.moveaxis(lowerCAmelCase_		,				0		,				-1							)							) for x in image_inputs]

        return image_inputs



 def      UpperCAmelCase							(					self							)  ->					str:
        _A        =  InstructBlipProcessor(
            tokenizer=self.get_tokenizer()		,				image_processor=self.get_image_processor()		,				qformer_tokenizer=self.get_qformer_tokenizer()		,				)
        processor.save_pretrained(self.tmpdirname							)

        _A        =  self.get_tokenizer(bos_token="""(BOS)"""		,				eos_token="""(EOS)"""							)
        _A        =  self.get_image_processor(do_normalize=lowerCAmelCase_		,				padding_value=1.0							)

        _A        =  InstructBlipProcessor.from_pretrained(
            self.tmpdirname		,				bos_token="""(BOS)"""		,				eos_token="""(EOS)"""		,				do_normalize=lowerCAmelCase_		,				padding_value=1.0							)

        self.assertEqual(processor.tokenizer.get_vocab()		,				tokenizer_add_kwargs.get_vocab()							)
        self.assertIsInstance(processor.tokenizer		,				lowerCAmelCase_							)

        self.assertEqual(processor.image_processor.to_json_string()		,				image_processor_add_kwargs.to_json_string()							)
        self.assertIsInstance(processor.image_processor		,				lowerCAmelCase_							)
        self.assertIsInstance(processor.qformer_tokenizer		,				lowerCAmelCase_							)



 def      UpperCAmelCase							(					self							)  ->					int:
        _A        =  self.get_image_processor()
        _A        =  self.get_tokenizer()
        _A        =  self.get_qformer_tokenizer()

        _A        =  InstructBlipProcessor(
            tokenizer=lowerCAmelCase_		,				image_processor=lowerCAmelCase_		,				qformer_tokenizer=lowerCAmelCase_							)

        _A        =  self.prepare_image_inputs()

        _A        =  image_processor(lowerCAmelCase_		,				return_tensors="""np"""							)
        _A        =  processor(images=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							)  ->					Dict:
        _A        =  self.get_image_processor()
        _A        =  self.get_tokenizer()
        _A        =  self.get_qformer_tokenizer()

        _A        =  InstructBlipProcessor(
            tokenizer=lowerCAmelCase_		,				image_processor=lowerCAmelCase_		,				qformer_tokenizer=lowerCAmelCase_							)

        _A        =  """lower newer"""

        _A        =  processor(text=lowerCAmelCase_							)

        _A        =  tokenizer(lowerCAmelCase_		,				return_token_type_ids=lowerCAmelCase_							)
        _A        =  qformer_tokenizer(lowerCAmelCase_		,				return_token_type_ids=lowerCAmelCase_							)

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

        for key in encoded_tokens_qformer.keys():
               self.assertListEqual(encoded_tokens_qformer[key]		,				encoded_processor["""qformer_""" + key]							)



 def      UpperCAmelCase							(					self							)  ->					List[str]:
        _A        =  self.get_image_processor()
        _A        =  self.get_tokenizer()
        _A        =  self.get_qformer_tokenizer()

        _A        =  InstructBlipProcessor(
            tokenizer=lowerCAmelCase_		,				image_processor=lowerCAmelCase_		,				qformer_tokenizer=lowerCAmelCase_							)

        _A        =  """lower newer"""
        _A        =  self.prepare_image_inputs()

        _A        =  processor(text=lowerCAmelCase_		,				images=lowerCAmelCase_							)

        self.assertListEqual(
            list(inputs.keys()							)		,				["""input_ids""", """attention_mask""", """qformer_input_ids""", """qformer_attention_mask""", """pixel_values"""]		,				)

        # test if it raises when no input is passed
        with pytest.raises(lowerCAmelCase_							):
               processor()



 def      UpperCAmelCase							(					self							)  ->					int:
        _A        =  self.get_image_processor()
        _A        =  self.get_tokenizer()
        _A        =  self.get_qformer_tokenizer()

        _A        =  InstructBlipProcessor(
            tokenizer=lowerCAmelCase_		,				image_processor=lowerCAmelCase_		,				qformer_tokenizer=lowerCAmelCase_							)

        _A        =  [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

        _A        =  processor.batch_decode(lowerCAmelCase_							)
        _A        =  tokenizer.batch_decode(lowerCAmelCase_							)

        self.assertListEqual(lowerCAmelCase_		,				lowerCAmelCase_							)



 def      UpperCAmelCase							(					self							)  ->					Optional[Any]:
        _A        =  self.get_image_processor()
        _A        =  self.get_tokenizer()
        _A        =  self.get_qformer_tokenizer()

        _A        =  InstructBlipProcessor(
            tokenizer=lowerCAmelCase_		,				image_processor=lowerCAmelCase_		,				qformer_tokenizer=lowerCAmelCase_							)

        _A        =  """lower newer"""
        _A        =  self.prepare_image_inputs()

        _A        =  processor(text=lowerCAmelCase_		,				images=lowerCAmelCase_							)

        self.assertListEqual(
            list(inputs.keys()							)		,				["""input_ids""", """attention_mask""", """qformer_input_ids""", """qformer_attention_mask""", """pixel_values"""]		,				)




