





import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
	from huggingface_hub import snapshot_download
	from pyctcdecode import BeamSearchDecoderCTC

	from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
	from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
	from transformers import WavaVecaForCTC




@require_pyctcdecode
class 	UpperCamelCase_							(  unittest.TestCase     ):


							def   _snake_case	(				self     :Dict		) -> int:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   """| <pad> <unk> <s> </s> a b c d e f g h i j k""".split()
									SCREAMING_SNAKE_CASE__								=   dict(zip(__A				,					range(len(__A		)		)		)		)

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

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

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

									# load decoder from hub
									SCREAMING_SNAKE_CASE__								=   """hf-internal-testing/ngram-beam-search-decoder"""


							def   _snake_case	(				self     :List[Any]				,					**__A     :Optional[Any]		) -> Optional[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.add_kwargs_tokens_map.copy()
									kwargs.update(__A		)
									return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname				,					**__A		)


							def   _snake_case	(				self     :int				,					**__A     :Union[str, Any]		) -> Tuple:



									"""simple docstring"""



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


							def   _snake_case	(				self     :List[str]				,					**__A     :Union[str, Any]		) -> List[str]:



									"""simple docstring"""



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


							def   _snake_case	(				self     :List[Any]		) -> Dict:



									"""simple docstring"""



									shutil.rmtree(self.tmpdirname		)


							def   _snake_case	(				self     :List[str]		) -> Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.get_tokenizer()
									SCREAMING_SNAKE_CASE__								=   self.get_feature_extractor()
									SCREAMING_SNAKE_CASE__								=   self.get_decoder()

									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM(tokenizer=__A				,					feature_extractor=__A				,					decoder=__A		)

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

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

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

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


							def   _snake_case	(				self     :Any		) -> Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM(
									    tokenizer=self.get_tokenizer()				,					feature_extractor=self.get_feature_extractor()				,					decoder=self.get_decoder()		)
									processor.save_pretrained(self.tmpdirname		)

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

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


							def   _snake_case	(				self     :Any		) -> List[str]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.get_tokenizer()
									# add token to trigger raise
									tokenizer.add_tokens(["""xx"""]		)
									with self.assertRaisesRegex(__A				,					"""include"""		):
											WavaVecaProcessorWithLM(
											    tokenizer=__A				,					feature_extractor=self.get_feature_extractor()				,					decoder=self.get_decoder()		)


							def   _snake_case	(				self     :List[Any]		) -> Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.get_feature_extractor()
									SCREAMING_SNAKE_CASE__								=   self.get_tokenizer()
									SCREAMING_SNAKE_CASE__								=   self.get_decoder()

									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM(tokenizer=__A				,					feature_extractor=__A				,					decoder=__A		)

									SCREAMING_SNAKE_CASE__								=   floats_list((3, 1000)		)

									SCREAMING_SNAKE_CASE__								=   feature_extractor(__A				,					return_tensors="""np"""		)
									SCREAMING_SNAKE_CASE__								=   processor(__A				,					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   _snake_case	(				self     :Tuple		) -> Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.get_feature_extractor()
									SCREAMING_SNAKE_CASE__								=   self.get_tokenizer()
									SCREAMING_SNAKE_CASE__								=   self.get_decoder()

									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM(tokenizer=__A				,					feature_extractor=__A				,					decoder=__A		)

									SCREAMING_SNAKE_CASE__								=   """This is a test string"""

									SCREAMING_SNAKE_CASE__								=   processor(text=__A		)

									SCREAMING_SNAKE_CASE__								=   tokenizer(__A		)

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


							def   _snake_case	(				self     :Optional[Any]				,					__A     :List[Any]=(2, 10, 16)				,					__A     :Any=77		) -> str:



									"""simple docstring"""



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


							def   _snake_case	(				self     :List[str]		) -> str:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.get_feature_extractor()
									SCREAMING_SNAKE_CASE__								=   self.get_tokenizer()
									SCREAMING_SNAKE_CASE__								=   self.get_decoder()

									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM(tokenizer=__A				,					feature_extractor=__A				,					decoder=__A		)

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

									SCREAMING_SNAKE_CASE__								=   processor.decode(__A		)

									SCREAMING_SNAKE_CASE__								=   decoder.decode_beams(__A		)[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   _snake_case	(				self     :List[Any]				,					__A     :str		) -> Optional[int]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.get_feature_extractor()
									SCREAMING_SNAKE_CASE__								=   self.get_tokenizer()
									SCREAMING_SNAKE_CASE__								=   self.get_decoder()

									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM(tokenizer=__A				,					feature_extractor=__A				,					decoder=__A		)

									SCREAMING_SNAKE_CASE__								=   self._get_dummy_logits()

									# note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
									#       otherwise, the LM won't be available to the pool's sub-processes.
									# manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
									if pool_context is None:
											SCREAMING_SNAKE_CASE__								=   processor.batch_decode(__A		)
									else:
											with get_context(__A		).Pool() as pool:
													SCREAMING_SNAKE_CASE__								=   processor.batch_decode(__A				,					__A		)

									SCREAMING_SNAKE_CASE__								=   list(__A		)

									with get_context("""fork"""		).Pool() as p:
											SCREAMING_SNAKE_CASE__								=   decoder.decode_beams_batch(__A				,					__A		)

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

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


							def   _snake_case	(				self     :List[str]		) -> str:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.get_feature_extractor()
									SCREAMING_SNAKE_CASE__								=   self.get_tokenizer()
									SCREAMING_SNAKE_CASE__								=   self.get_decoder()

									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM(tokenizer=__A				,					feature_extractor=__A				,					decoder=__A		)

									SCREAMING_SNAKE_CASE__								=   self._get_dummy_logits()

									SCREAMING_SNAKE_CASE__								=   15
									SCREAMING_SNAKE_CASE__								=   -2_0.0
									SCREAMING_SNAKE_CASE__								=   -4.0

									SCREAMING_SNAKE_CASE__								=   processor.batch_decode(
									    __A				,					beam_width=__A				,					beam_prune_logp=__A				,					token_min_logp=__A				,					)
									SCREAMING_SNAKE_CASE__								=   decoded_processor_out.text

									SCREAMING_SNAKE_CASE__								=   list(__A		)

									with get_context("""fork"""		).Pool() as pool:
											SCREAMING_SNAKE_CASE__								=   decoder.decode_beams_batch(
											    __A				,					__A				,					beam_width=__A				,					beam_prune_logp=__A				,					token_min_logp=__A				,					)

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

									self.assertListEqual(__A				,					__A		)
									self.assertListEqual(["""</s> <s> <s>""", """<s> <s> <s>"""]				,					__A		)

									self.assertTrue(np.array_equal(__A				,					decoded_processor_out.logit_score		)		)
									self.assertTrue(np.allclose([-2_0.0_5_4, -1_8.4_4_7]				,					__A				,					atol=1E-3		)		)

									self.assertTrue(np.array_equal(__A				,					decoded_processor_out.lm_score		)		)
									self.assertTrue(np.allclose([-1_5.5_5_4, -1_3.9_4_7_4]				,					__A				,					atol=1E-3		)		)


							def   _snake_case	(				self     :Any		) -> Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.get_feature_extractor()
									SCREAMING_SNAKE_CASE__								=   self.get_tokenizer()
									SCREAMING_SNAKE_CASE__								=   self.get_decoder()

									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM(tokenizer=__A				,					feature_extractor=__A				,					decoder=__A		)

									SCREAMING_SNAKE_CASE__								=   self._get_dummy_logits()

									SCREAMING_SNAKE_CASE__								=   2.0
									SCREAMING_SNAKE_CASE__								=   5.0
									SCREAMING_SNAKE_CASE__								=   -2_0.0
									SCREAMING_SNAKE_CASE__								=   True

									SCREAMING_SNAKE_CASE__								=   processor.batch_decode(
									    __A				,					alpha=__A				,					beta=__A				,					unk_score_offset=__A				,					lm_score_boundary=__A				,					)
									SCREAMING_SNAKE_CASE__								=   decoded_processor_out.text

									SCREAMING_SNAKE_CASE__								=   list(__A		)
									decoder.reset_params(
									    alpha=__A				,					beta=__A				,					unk_score_offset=__A				,					lm_score_boundary=__A				,					)

									with get_context("""fork"""		).Pool() as pool:
											SCREAMING_SNAKE_CASE__								=   decoder.decode_beams_batch(
											    __A				,					__A				,					)

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

									self.assertListEqual(__A				,					__A		)
									self.assertListEqual(["""<s> </s> <s> </s> </s>""", """</s> </s> <s> </s> </s>"""]				,					__A		)
									SCREAMING_SNAKE_CASE__								=   processor.decoder.model_container[processor.decoder._model_key]
									self.assertEqual(lm_model.alpha				,					2.0		)
									self.assertEqual(lm_model.beta				,					5.0		)
									self.assertEqual(lm_model.unk_score_offset				,					-2_0.0		)
									self.assertEqual(lm_model.score_boundary				,					__A		)


							def   _snake_case	(				self     :Any		) -> Tuple:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""		)

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

									SCREAMING_SNAKE_CASE__								=   os.listdir(__A		)
									SCREAMING_SNAKE_CASE__								=   ["""alphabet.json""", """language_model"""]

									downloaded_decoder_files.sort()
									expected_decoder_files.sort()

									# test that only decoder relevant files from
									# https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
									# are downloaded and none of the rest (e.g. README.md, ...)
									self.assertListEqual(__A				,					__A		)


							def   _snake_case	(				self     :Tuple		) -> str:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   snapshot_download("""hf-internal-testing/processor_with_lm"""		)

									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM.from_pretrained(__A		)

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

									SCREAMING_SNAKE_CASE__								=   os.listdir(__A		)
									SCREAMING_SNAKE_CASE__								=   os.listdir(__A		)

									local_decoder_files.sort()
									expected_decoder_files.sort()

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


							def   _snake_case	(				self     :int		) -> List[str]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""		)
									SCREAMING_SNAKE_CASE__								=   AutoProcessor.from_pretrained("""hf-internal-testing/processor_with_lm"""		)

									SCREAMING_SNAKE_CASE__								=   floats_list((3, 1000)		)

									SCREAMING_SNAKE_CASE__								=   processor_wavaveca(__A				,					return_tensors="""np"""		)
									SCREAMING_SNAKE_CASE__								=   processor_auto(__A				,					return_tensors="""np"""		)

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

									SCREAMING_SNAKE_CASE__								=   self._get_dummy_logits()

									SCREAMING_SNAKE_CASE__								=   processor_wavaveca.batch_decode(__A		)
									SCREAMING_SNAKE_CASE__								=   processor_auto.batch_decode(__A		)

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


							def   _snake_case	(				self     :Optional[int]		) -> int:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.get_feature_extractor()
									SCREAMING_SNAKE_CASE__								=   self.get_tokenizer()
									SCREAMING_SNAKE_CASE__								=   self.get_decoder()

									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM(tokenizer=__A				,					feature_extractor=__A				,					decoder=__A		)

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


							@staticmethod
							def   _snake_case	(				__A     :Any				,					__A     :Dict		) -> List[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [d[key] for d in offsets]
									return retrieved_list


							def   _snake_case	(				self     :List[Any]		) -> Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""		)
									SCREAMING_SNAKE_CASE__								=   self._get_dummy_logits()[0]

									SCREAMING_SNAKE_CASE__								=   processor.decode(__A				,					output_word_offsets=__A		)
									# 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(__A				,					__A		)		)

									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   _snake_case	(				self     :str		) -> Tuple:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""		)
									SCREAMING_SNAKE_CASE__								=   self._get_dummy_logits()

									SCREAMING_SNAKE_CASE__								=   processor.batch_decode(__A				,					output_word_offsets=__A		)

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

									self.assertListEqual(
									    [""" """.join(self.get_from_offsets(__A				,					"""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   _snake_case	(				self     :int		) -> int:



									"""simple docstring"""



									import torch

									SCREAMING_SNAKE_CASE__								=   load_dataset("""common_voice"""				,					"""en"""				,					split="""train"""				,					streaming=__A		)
									SCREAMING_SNAKE_CASE__								=   ds.cast_column("""audio"""				,					datasets.Audio(sampling_rate=1_6000		)		)
									SCREAMING_SNAKE_CASE__								=   iter(__A		)
									SCREAMING_SNAKE_CASE__								=   next(__A		)

									SCREAMING_SNAKE_CASE__								=   AutoProcessor.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""		)
									SCREAMING_SNAKE_CASE__								=   WavaVecaForCTC.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""		)

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

									with torch.no_grad():
											SCREAMING_SNAKE_CASE__								=   model(__A		).logits.cpu().numpy()

									SCREAMING_SNAKE_CASE__								=   processor.decode(logits[0]				,					output_word_offsets=__A		)

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

									SCREAMING_SNAKE_CASE__								=   """WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL"""

									# output words
									self.assertEqual(""" """.join(self.get_from_offsets(__A				,					"""word"""		)		)				,					__A		)
									self.assertEqual(""" """.join(self.get_from_offsets(__A				,					"""word"""		)		)				,					output.text		)

									# output times
									SCREAMING_SNAKE_CASE__								=   torch.tensor(self.get_from_offsets(__A				,					"""start_time"""		)		)
									SCREAMING_SNAKE_CASE__								=   torch.tensor(self.get_from_offsets(__A				,					"""end_time"""		)		)

									# fmt: off
									SCREAMING_SNAKE_CASE__								=   torch.tensor([1.4_1_9_9, 1.6_5_9_9, 2.2_5_9_9, 3.0, 3.2_4, 3.5_9_9_9, 3.7_9_9_9, 4.0_9_9_9, 4.2_6, 4.9_4, 5.2_8, 5.6_5_9_9, 5.7_8, 5.9_4, 6.3_2, 6.5_3_9_9, 6.6_5_9_9]		)
									SCREAMING_SNAKE_CASE__								=   torch.tensor([1.5_3_9_9, 1.8_9_9_9, 2.9, 3.1_6, 3.5_3_9_9, 3.7_2, 4.0_1_9_9, 4.1_7_9_9, 4.7_6, 5.1_5_9_9, 5.5_5_9_9, 5.6_9_9_9, 5.8_6, 6.1_9_9_9, 6.3_8, 6.6_1_9_9, 6.9_4]		)
									# fmt: on

									self.assertTrue(torch.allclose(__A				,					__A				,					atol=0.0_1		)		)
									self.assertTrue(torch.allclose(__A				,					__A				,					atol=0.0_1		)		)






import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available




class 	UpperCamelCase_							(  UpperCamelCase__     ):
							lowerCamelCase_							     =       ["image_processor", "tokenizer"]
							lowerCamelCase_							     =       "OwlViTImageProcessor"
							lowerCamelCase_							     =       ("CLIPTokenizer", "CLIPTokenizerFast")


							def __init__(				self     :Optional[Any]				,					__A     :int=None				,					__A     :Optional[int]=None				,					**__A     :str		) -> Optional[int]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   None
									if "feature_extractor" in kwargs:
											warnings.warn(
											    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
											    """ instead."""				,					__A				,					)
											SCREAMING_SNAKE_CASE__								=   kwargs.pop("""feature_extractor"""		)

									SCREAMING_SNAKE_CASE__								=   image_processor if image_processor is not None else feature_extractor
									if image_processor is None:
											raise ValueError("""You need to specify an `image_processor`."""		)
									if tokenizer is None:
											raise ValueError("""You need to specify a `tokenizer`."""		)

									super().__init__(__A				,					__A		)


							def __call__(				self     :str				,					__A     :Dict=None				,					__A     :List[str]=None				,					__A     :str=None				,					__A     :Optional[int]="max_length"				,					__A     :Tuple="np"				,					**__A     :int		) -> Tuple:



									"""simple docstring"""



									if text is None and query_images is None and images is None:
											raise ValueError(
											    """You have to specify at least one text or query image or image. All three cannot be none."""		)

									if text is not None:
											if isinstance(__A				,					__A		) or (isinstance(__A				,					__A		) and not isinstance(text[0]				,					__A		)):
													SCREAMING_SNAKE_CASE__								=   [self.tokenizer(__A				,					padding=__A				,					return_tensors=__A				,					**__A		)]

											elif isinstance(__A				,					__A		) and isinstance(text[0]				,					__A		):
													SCREAMING_SNAKE_CASE__								=   []

													# Maximum number of queries across batch
													SCREAMING_SNAKE_CASE__								=   max([len(__A		) for t in text]		)

													# Pad all batch samples to max number of text queries
													for t in text:
															if len(__A		) != max_num_queries:
																	SCREAMING_SNAKE_CASE__								=   t + [""" """] * (max_num_queries - len(__A		))

															SCREAMING_SNAKE_CASE__								=   self.tokenizer(__A				,					padding=__A				,					return_tensors=__A				,					**__A		)
															encodings.append(__A		)
											else:
													raise TypeError("""Input text should be a string, a list of strings or a nested list of strings"""		)

											if return_tensors == "np":
													SCREAMING_SNAKE_CASE__								=   np.concatenate([encoding["""input_ids"""] for encoding in encodings]				,					axis=0		)
													SCREAMING_SNAKE_CASE__								=   np.concatenate([encoding["""attention_mask"""] for encoding in encodings]				,					axis=0		)

											elif return_tensors == "jax" and is_flax_available():
													import jax.numpy as jnp

													SCREAMING_SNAKE_CASE__								=   jnp.concatenate([encoding["""input_ids"""] for encoding in encodings]				,					axis=0		)
													SCREAMING_SNAKE_CASE__								=   jnp.concatenate([encoding["""attention_mask"""] for encoding in encodings]				,					axis=0		)

											elif return_tensors == "pt" and is_torch_available():
													import torch

													SCREAMING_SNAKE_CASE__								=   torch.cat([encoding["""input_ids"""] for encoding in encodings]				,					dim=0		)
													SCREAMING_SNAKE_CASE__								=   torch.cat([encoding["""attention_mask"""] for encoding in encodings]				,					dim=0		)

											elif return_tensors == "tf" and is_tf_available():
													import tensorflow as tf

													SCREAMING_SNAKE_CASE__								=   tf.stack([encoding["""input_ids"""] for encoding in encodings]				,					axis=0		)
													SCREAMING_SNAKE_CASE__								=   tf.stack([encoding["""attention_mask"""] for encoding in encodings]				,					axis=0		)

											else:
													raise ValueError("""Target return tensor type could not be returned"""		)

											SCREAMING_SNAKE_CASE__								=   BatchEncoding()
											SCREAMING_SNAKE_CASE__								=   input_ids
											SCREAMING_SNAKE_CASE__								=   attention_mask

									if query_images is not None:
											SCREAMING_SNAKE_CASE__								=   BatchEncoding()
											SCREAMING_SNAKE_CASE__								=   self.image_processor(
											    __A				,					return_tensors=__A				,					**__A		).pixel_values
											SCREAMING_SNAKE_CASE__								=   query_pixel_values

									if images is not None:
											SCREAMING_SNAKE_CASE__								=   self.image_processor(__A				,					return_tensors=__A				,					**__A		)

									if text is not None and images is not None:
											SCREAMING_SNAKE_CASE__								=   image_features.pixel_values
											return encoding
									elif query_images is not None and images is not None:
											SCREAMING_SNAKE_CASE__								=   image_features.pixel_values
											return encoding
									elif text is not None or query_images is not None:
											return encoding
									else:
											return BatchEncoding(data=dict(**__A		)				,					tensor_type=__A		)


							def   _snake_case	(				self     :List[Any]				,					*__A     :Dict				,					**__A     :Dict		) -> Optional[int]:



									"""simple docstring"""



									return self.image_processor.post_process(*__A				,					**__A		)


							def   _snake_case	(				self     :Optional[int]				,					*__A     :Dict				,					**__A     :List[str]		) -> Optional[Any]:



									"""simple docstring"""



									return self.image_processor.post_process_object_detection(*__A				,					**__A		)


							def   _snake_case	(				self     :str				,					*__A     :List[str]				,					**__A     :Union[str, Any]		) -> Any:



									"""simple docstring"""



									return self.image_processor.post_process_image_guided_detection(*__A				,					**__A		)


							def   _snake_case	(				self     :Dict				,					*__A     :List[str]				,					**__A     :List[str]		) -> int:



									"""simple docstring"""



									return self.tokenizer.batch_decode(*__A				,					**__A		)


							def   _snake_case	(				self     :Dict				,					*__A     :Dict				,					**__A     :List[str]		) -> str:



									"""simple docstring"""



									return self.tokenizer.decode(*__A				,					**__A		)


							@property
							def   _snake_case	(				self     :List[Any]		) -> Optional[int]:



									"""simple docstring"""



									warnings.warn(
									    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""				,					__A				,					)
									return self.image_processor_class


							@property
							def   _snake_case	(				self     :Any		) -> Optional[Any]:



									"""simple docstring"""



									warnings.warn(
									    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""				,					__A				,					)
									return self.image_processor