






"""simple docstring"""





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

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

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

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


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

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

if is_torch_available():
							from transformers import WavaVecaForCTC



@require_pyctcdecode
class       __A		(   unittest.TestCase			):

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							Optional[Any]:
											__magic_name__:	List[str]							=       """| <pad> <unk> <s> </s> a b c d e f g h i j k""".split()
											__magic_name__:	List[Any]							=       dict(zip(__snake_case     ,		range(len(__snake_case					)					)					)					)

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

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

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

											# load decoder from hub
											__magic_name__:	Dict							=       """hf-internal-testing/ngram-beam-search-decoder"""

					def   lowerCamelCase__   (			self  :      Any     ,		**__snake_case  :      str					)			->							Optional[int]:
											__magic_name__:	Union[str, Any]							=       self.add_kwargs_tokens_map.copy()
											kwargs.update(__snake_case					)
											return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname     ,		**__snake_case					)

					def   lowerCamelCase__   (			self  :      str     ,		**__snake_case  :      int					)			->							Dict:
											return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname     ,		**__snake_case					)

					def   lowerCamelCase__   (			self  :      int     ,		**__snake_case  :      List[str]					)			->							int:
											return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name     ,		**__snake_case					)

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

					def   lowerCamelCase__   (			self  :      List[Any]					)			->							Optional[Any]:
											__magic_name__:	Dict							=       self.get_tokenizer()
											__magic_name__:	Any							=       self.get_feature_extractor()
											__magic_name__:	Tuple							=       self.get_decoder()

											__magic_name__:	Tuple							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											processor.save_pretrained(self.tmpdirname					)
											__magic_name__:	Dict							=       WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname					)

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

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

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

					def   lowerCamelCase__   (			self  :      Any					)			->							Tuple:
											__magic_name__:	Union[str, Any]							=       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
											__magic_name__:	int							=       WavaVecaProcessorWithLM.from_pretrained(
											    self.tmpdirname     ,		alpha=5.0     ,		beta=3.0     ,		score_boundary=-7.0     ,		unk_score_offset=3					)

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

					def   lowerCamelCase__   (			self  :      Optional[int]					)			->							Optional[Any]:
											__magic_name__:	Union[str, Any]							=       self.get_tokenizer()
											# add token to trigger raise
											tokenizer.add_tokens(["""xx"""]					)
											with self.assertRaisesRegex(__snake_case     ,		"""include"""					):
																	WavaVecaProcessorWithLM(
																	    tokenizer=__snake_case     ,		feature_extractor=self.get_feature_extractor()     ,		decoder=self.get_decoder()					)

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							int:
											__magic_name__:	int							=       self.get_feature_extractor()
											__magic_name__:	Optional[Any]							=       self.get_tokenizer()
											__magic_name__:	List[Any]							=       self.get_decoder()

											__magic_name__:	int							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

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

											__magic_name__:	List[str]							=       feature_extractor(__snake_case     ,		return_tensors="""np"""					)
											__magic_name__:	Tuple							=       processor(__snake_case     ,		return_tensors="""np"""					)

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

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							List[str]:
											__magic_name__:	Tuple							=       self.get_feature_extractor()
											__magic_name__:	List[str]							=       self.get_tokenizer()
											__magic_name__:	str							=       self.get_decoder()

											__magic_name__:	Tuple							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											__magic_name__:	Optional[int]							=       """This is a test string"""

											__magic_name__:	List[str]							=       processor(text=__snake_case					)

											__magic_name__:	Tuple							=       tokenizer(__snake_case					)

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

					def   lowerCamelCase__   (			self  :      int     ,		__snake_case  :      List[str]=(2, 1_0, 1_6)     ,		__snake_case  :      List[Any]=7_7					)			->							Dict:
											np.random.seed(__snake_case					)
											return np.random.rand(*__snake_case					)

					def   lowerCamelCase__   (			self  :      Any					)			->							Any:
											__magic_name__:	int							=       self.get_feature_extractor()
											__magic_name__:	Tuple							=       self.get_tokenizer()
											__magic_name__:	Any							=       self.get_decoder()

											__magic_name__:	Tuple							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

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

											__magic_name__:	str							=       processor.decode(__snake_case					)

											__magic_name__:	Optional[int]							=       decoder.decode_beams(__snake_case					)[0]

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

					@parameterized.expand([[None], ["""fork"""], ["""spawn"""]]					)
					def   lowerCamelCase__   (			self  :      int     ,		__snake_case  :      Dict					)			->							Any:
											__magic_name__:	int							=       self.get_feature_extractor()
											__magic_name__:	List[Any]							=       self.get_tokenizer()
											__magic_name__:	int							=       self.get_decoder()

											__magic_name__:	Optional[Any]							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											__magic_name__:	Optional[int]							=       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:
																	__magic_name__:	Optional[int]							=       processor.batch_decode(__snake_case					)
											else:
																	with get_context(__snake_case					).Pool() as pool:
																							__magic_name__:	Any							=       processor.batch_decode(__snake_case     ,		__snake_case					)

											__magic_name__:	Dict							=       list(__snake_case					)

											with get_context("""fork"""					).Pool() as p:
																	__magic_name__:	List[str]							=       decoder.decode_beams_batch(__snake_case     ,		__snake_case					)

											__magic_name__,			__magic_name__,			__magic_name__:	Optional[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(__snake_case     ,		decoded_processor.text					)
											self.assertListEqual(["""<s> <s> </s>""", """<s> <s> <s>"""]     ,		decoded_processor.text					)
											self.assertListEqual(__snake_case     ,		decoded_processor.logit_score					)
											self.assertListEqual(__snake_case     ,		decoded_processor.lm_score					)

					def   lowerCamelCase__   (			self  :      List[Any]					)			->							List[Any]:
											__magic_name__:	List[str]							=       self.get_feature_extractor()
											__magic_name__:	Optional[Any]							=       self.get_tokenizer()
											__magic_name__:	Optional[int]							=       self.get_decoder()

											__magic_name__:	Dict							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											__magic_name__:	str							=       self._get_dummy_logits()

											__magic_name__:	Dict							=       1_5
											__magic_name__:	int							=       -20.0
											__magic_name__:	int							=       -4.0

											__magic_name__:	Dict							=       processor.batch_decode(
											    __snake_case     ,		beam_width=__snake_case     ,		beam_prune_logp=__snake_case     ,		token_min_logp=__snake_case     ,		)
											__magic_name__:	Optional[int]							=       decoded_processor_out.text

											__magic_name__:	Union[str, Any]							=       list(__snake_case					)

											with get_context("""fork"""					).Pool() as pool:
																	__magic_name__:	str							=       decoder.decode_beams_batch(
																	    __snake_case     ,		__snake_case     ,		beam_width=__snake_case     ,		beam_prune_logp=__snake_case     ,		token_min_logp=__snake_case     ,		)

											__magic_name__:	Any							=       [d[0][0] for d in decoded_decoder_out]
											__magic_name__:	Optional[int]							=       [d[0][2] for d in decoded_decoder_out]
											__magic_name__:	Optional[Any]							=       [d[0][3] for d in decoded_decoder_out]

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

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

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

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							int:
											__magic_name__:	int							=       self.get_feature_extractor()
											__magic_name__:	Any							=       self.get_tokenizer()
											__magic_name__:	Union[str, Any]							=       self.get_decoder()

											__magic_name__:	str							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											__magic_name__:	Any							=       self._get_dummy_logits()

											__magic_name__:	Union[str, Any]							=       2.0
											__magic_name__:	Optional[Any]							=       5.0
											__magic_name__:	Optional[Any]							=       -20.0
											__magic_name__:	List[str]							=       True

											__magic_name__:	List[Any]							=       processor.batch_decode(
											    __snake_case     ,		alpha=__snake_case     ,		beta=__snake_case     ,		unk_score_offset=__snake_case     ,		lm_score_boundary=__snake_case     ,		)
											__magic_name__:	Union[str, Any]							=       decoded_processor_out.text

											__magic_name__:	Union[str, Any]							=       list(__snake_case					)
											decoder.reset_params(
											    alpha=__snake_case     ,		beta=__snake_case     ,		unk_score_offset=__snake_case     ,		lm_score_boundary=__snake_case     ,		)

											with get_context("""fork"""					).Pool() as pool:
																	__magic_name__:	str							=       decoder.decode_beams_batch(
																	    __snake_case     ,		__snake_case     ,		)

											__magic_name__:	List[str]							=       [d[0][0] for d in decoded_decoder_out]

											self.assertListEqual(__snake_case     ,		__snake_case					)
											self.assertListEqual(["""<s> </s> <s> </s> </s>""", """</s> </s> <s> </s> </s>"""]     ,		__snake_case					)
											__magic_name__:	List[str]							=       processor.decoder.model_container[processor.decoder._model_key]
											self.assertEqual(lm_model.alpha     ,		2.0					)
											self.assertEqual(lm_model.beta     ,		5.0					)
											self.assertEqual(lm_model.unk_score_offset     ,		-20.0					)
											self.assertEqual(lm_model.score_boundary     ,		__snake_case					)

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							List[Any]:
											__magic_name__:	List[Any]							=       WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""					)

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

											__magic_name__:	Optional[int]							=       os.listdir(__snake_case					)
											__magic_name__:	Union[str, Any]							=       ["""alphabet.json""", """language_model"""]

											downloaded_decoder_files.sort()
											expected_decoder_files.sort()

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

					def   lowerCamelCase__   (			self  :      Any					)			->							Any:
											__magic_name__:	int							=       snapshot_download("""hf-internal-testing/processor_with_lm"""					)

											__magic_name__:	List[Any]							=       WavaVecaProcessorWithLM.from_pretrained(__snake_case					)

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

											__magic_name__:	str							=       os.listdir(__snake_case					)
											__magic_name__:	Tuple							=       os.listdir(__snake_case					)

											local_decoder_files.sort()
											expected_decoder_files.sort()

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

					def   lowerCamelCase__   (			self  :      Optional[int]					)			->							int:
											__magic_name__:	List[Any]							=       WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""					)
											__magic_name__:	List[str]							=       AutoProcessor.from_pretrained("""hf-internal-testing/processor_with_lm"""					)

											__magic_name__:	List[str]							=       floats_list((3, 1_0_0_0)					)

											__magic_name__:	Tuple							=       processor_wavaveca(__snake_case     ,		return_tensors="""np"""					)
											__magic_name__:	Optional[Any]							=       processor_auto(__snake_case     ,		return_tensors="""np"""					)

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

											__magic_name__:	int							=       self._get_dummy_logits()

											__magic_name__:	List[Any]							=       processor_wavaveca.batch_decode(__snake_case					)
											__magic_name__:	Union[str, Any]							=       processor_auto.batch_decode(__snake_case					)

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

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							str:
											__magic_name__:	Optional[int]							=       self.get_feature_extractor()
											__magic_name__:	Any							=       self.get_tokenizer()
											__magic_name__:	Dict							=       self.get_decoder()

											__magic_name__:	List[str]							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

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

					@staticmethod
					def   lowerCamelCase__   (			__snake_case  :      Optional[int]     ,		__snake_case  :      int					)			->							int:
											__magic_name__:	Any							=       [d[key] for d in offsets]
											return retrieved_list

					def   lowerCamelCase__   (			self  :      str					)			->							Union[str, Any]:
											__magic_name__:	Tuple							=       WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""					)
											__magic_name__:	Tuple							=       self._get_dummy_logits()[0]

											__magic_name__:	List[Any]							=       processor.decode(__snake_case     ,		output_word_offsets=__snake_case					)
											# check Wav2Vec2CTCTokenizerOutput keys for word
											self.assertEqual(len(outputs.keys()					)     ,		4					)
											self.assertTrue("""text""" in outputs					)
											self.assertTrue("""word_offsets""" in outputs					)
											self.assertTrue(isinstance(__snake_case     ,		__snake_case					)					)

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

					def   lowerCamelCase__   (			self  :      Optional[int]					)			->							Dict:
											__magic_name__:	Optional[int]							=       WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""					)
											__magic_name__:	Optional[int]							=       self._get_dummy_logits()

											__magic_name__:	Any							=       processor.batch_decode(__snake_case     ,		output_word_offsets=__snake_case					)

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

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






					@slow
					@require_torch
					@require_torchaudio
					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							int:
											import torch

											__magic_name__:	List[Any]							=       load_dataset("""common_voice"""     ,		"""en"""     ,		split="""train"""     ,		streaming=__snake_case					)
											__magic_name__:	Dict							=       ds.cast_column("""audio"""     ,		datasets.Audio(sampling_rate=1_6_0_0_0					)					)
											__magic_name__:	Any							=       iter(__snake_case					)
											__magic_name__:	Optional[int]							=       next(__snake_case					)

											__magic_name__:	Optional[int]							=       AutoProcessor.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""					)
											__magic_name__:	Tuple							=       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
											__magic_name__:	List[str]							=       processor(sample["""audio"""]["""array"""]     ,		return_tensors="""pt"""					).input_values

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

											__magic_name__:	Optional[Any]							=       processor.decode(logits[0]     ,		output_word_offsets=__snake_case					)

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

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

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

											# output times
											__magic_name__:	Dict							=       torch.tensor(self.get_from_offsets(__snake_case     ,		"""start_time"""					)					)
											__magic_name__:	Optional[Any]							=       torch.tensor(self.get_from_offsets(__snake_case     ,		"""end_time"""					)					)

											# fmt: off
											__magic_name__:	Tuple							=       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]					)
											__magic_name__:	int							=       torch.tensor([1.5399, 1.8999, 2.9, 3.16, 3.5399, 3.72, 4.0199, 4.1799, 4.76, 5.1599, 5.5599, 5.6999, 5.86, 6.1999, 6.38, 6.6199, 6.94]					)
											# fmt: on

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





"""simple docstring"""





import inspect
import unittest

from transformers import ViTHybridConfig
from transformers.testing_utils import require_accelerate, require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
				import torch
				from torch import nn

				from transformers import ViTHybridForImageClassification, ViTHybridImageProcessor, ViTHybridModel
				from transformers.models.vit_hybrid.modeling_vit_hybrid import VIT_HYBRID_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
				from PIL import Image






class        _UpperCAmelCase:



		def __init__(   self							,						__a							,						__a=13							,						__a=64							,						__a=2							,						__a=3							,						__a=True							,						__a=True							,						__a=32							,						__a=5							,						__a=4							,						__a=37							,						__a="gelu"							,						__a=0.1							,						__a=0.1							,						__a=10							,						__a=0.02							,						__a=[1, 16, 4, 4]							,						__a=None							,						)			->			Optional[Any]:






							'''simple docstring'''
							_UpperCamelCase									=    parent
							_UpperCamelCase									=    batch_size
							_UpperCamelCase									=    image_size
							_UpperCamelCase									=    patch_size
							_UpperCamelCase									=    num_channels
							_UpperCamelCase									=    is_training
							_UpperCamelCase									=    use_labels
							_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									=    type_sequence_label_size
							_UpperCamelCase									=    initializer_range
							_UpperCamelCase									=    scope
							_UpperCamelCase									=    backbone_featmap_shape

							# in ViT hybrid, the seq length equals the number of patches + 1 (we add 1 for the [CLS] token)
							# the number of patches is based on the feature map of the backbone, which by default uses an output stride
							# of 32, which means that the feature map has a spatial resolution of 1/32 of the input image size
							_UpperCamelCase									=    (self.image_size // 32) ** 2
							_UpperCamelCase									=    num_patches + 1



		def 		UpperCAmelCase				(   self)			->			int:






							'''simple docstring'''
							_UpperCamelCase									=    floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size])

							_UpperCamelCase									=    None
							if self.use_labels:
												_UpperCamelCase									=    ids_tensor([self.batch_size]							,						self.type_sequence_label_size)

							_UpperCamelCase									=    self.get_config()

							return config, pixel_values, labels



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






							'''simple docstring'''
							_UpperCamelCase									=    {
							    '''global_padding''': '''same''',
							    '''layer_type''': '''bottleneck''',
							    '''depths''': [3, 4, 9],
							    '''out_features''': ['''stage1''', '''stage2''', '''stage3'''],
							    '''embedding_dynamic_padding''': True,
							    '''hidden_sizes''': [4, 8, 16, 32],
							    '''num_groups''': 2,
							}

							return ViTHybridConfig(
							    image_size=self.image_size							,						patch_size=self.patch_size							,						num_channels=self.num_channels							,						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							,						is_decoder=__a							,						initializer_range=self.initializer_range							,						backbone_featmap_shape=self.backbone_featmap_shape							,						backbone_config=__a							,						)



		def 		UpperCAmelCase				(   self							,						__a							,						__a							,						__a)			->			Optional[Any]:






							'''simple docstring'''
							_UpperCamelCase									=    ViTHybridModel(config=__a)
							model.to(__a)
							model.eval()
							_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)			->			Union[str, Any]:






							'''simple docstring'''
							_UpperCamelCase									=    self.type_sequence_label_size
							_UpperCamelCase									=    ViTHybridForImageClassification(__a)
							model.to(__a)
							model.eval()
							_UpperCamelCase									=    model(__a							,						labels=__a)
							self.parent.assertEqual(result.logits.shape							,						(self.batch_size, self.type_sequence_label_size))




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






							'''simple docstring'''
							_UpperCamelCase									=    self.prepare_config_and_inputs()
							_UpperCamelCase					,    _UpperCamelCase					,    _UpperCamelCase									=    config_and_inputs
							_UpperCamelCase									=    {'''pixel_values''': pixel_values}
							return config, inputs_dict






@require_torch
class        _UpperCAmelCase(       lowerCamelCase ,	lowerCamelCase ,	unittest.TestCase      ):
		lowercase__         = (ViTHybridModel, ViTHybridForImageClassification) if is_torch_available() else ()
		lowercase__         = (
		    {'feature-extraction': ViTHybridModel, 'image-classification': ViTHybridForImageClassification}
		    if is_torch_available()
		    else {}
		)
		lowercase__         = False
		lowercase__         = False
		lowercase__         = False



		def 		UpperCAmelCase				(   self)			->			Optional[Any]:






							'''simple docstring'''
							_UpperCamelCase									=    ViTHybridModelTester(self)
							_UpperCamelCase									=    ConfigTester(self							,						config_class=__a							,						has_text_modality=__a							,						hidden_size=37)



		def 		UpperCAmelCase				(   self)			->			Any:






							'''simple docstring'''
							self.config_tester.run_common_tests()



		@unittest.skip(reason='''ViT does not use inputs_embeds''')
		def 		UpperCAmelCase				(   self)			->			Dict:






							'''simple docstring'''
							pass



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






							'''simple docstring'''
							_UpperCamelCase					,    _UpperCamelCase									=    self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
												_UpperCamelCase									=    model_class(__a)
												self.assertIsInstance(model.get_input_embeddings()							,						(nn.Module))
												_UpperCamelCase									=    model.get_output_embeddings()
												self.assertTrue(x is None or isinstance(__a							,						nn.Linear))



		def 		UpperCAmelCase				(   self)			->			Tuple:






							'''simple docstring'''
							_UpperCamelCase					,    _UpperCamelCase									=    self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
												_UpperCamelCase									=    model_class(__a)
												_UpperCamelCase									=    inspect.signature(model.forward)
												# signature.parameters is an OrderedDict => so arg_names order is deterministic
												_UpperCamelCase									=    [*signature.parameters.keys()]

												_UpperCamelCase									=    ['''pixel_values''']
												self.assertListEqual(arg_names[:1]							,						__a)



		def 		UpperCAmelCase				(   self)			->			str:






							'''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()
							self.model_tester.create_and_check_for_image_classification(*__a)



		def 		UpperCAmelCase				(   self)			->			int:






							'''simple docstring'''
							_UpperCamelCase					,    _UpperCamelCase									=    self.model_tester.prepare_config_and_inputs_for_common()

							_UpperCamelCase									=    _config_zero_init(__a)
							for model_class in self.all_model_classes:
												_UpperCamelCase									=    model_class(config=__a)
												# Skip the check for the backbone
												for name, module in model.named_modules():
																	if module.__class__.__name__ == "ViTHybridPatchEmbeddings":
																						_UpperCamelCase									=    [F'''{name}.{key}''' for key in module.state_dict().keys()]
																						break

												for name, param in model.named_parameters():
																	if param.requires_grad:
																						if name in backbone_params:
																											continue
																						self.assertIn(
																						    ((param.data.mean() * 1e9).round() / 1e9).item()							,						[0.0, 1.0]							,						msg=F'''Parameter {name} of model {model_class} seems not properly initialized'''							,						)




		@slow
		def 		UpperCAmelCase				(   self)			->			Optional[Any]:






							'''simple docstring'''
							for model_name in VIT_HYBRID_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
												_UpperCamelCase									=    ViTHybridModel.from_pretrained(__a)
												self.assertIsNotNone(__a)





def  lowerCamelCase__						(     )				->						Tuple:


					"""simple docstring"""





					_UpperCamelCase									=    Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
					return image






@require_torch
@require_vision
class        _UpperCAmelCase(       unittest.TestCase      ):



		@cached_property
		def 		UpperCAmelCase				(   self)			->			Tuple:






							'''simple docstring'''
							return (
							    ViTHybridImageProcessor.from_pretrained(VIT_HYBRID_PRETRAINED_MODEL_ARCHIVE_LIST[0])
							    if is_vision_available()
							    else None
							)



		@slow
		def 		UpperCAmelCase				(   self)			->			Dict:






							'''simple docstring'''
							_UpperCamelCase									=    ViTHybridForImageClassification.from_pretrained(VIT_HYBRID_PRETRAINED_MODEL_ARCHIVE_LIST[0]).to(
							    __a)

							_UpperCamelCase									=    self.default_image_processor
							_UpperCamelCase									=    prepare_img()
							_UpperCamelCase									=    image_processor(images=__a							,						return_tensors='''pt''').to(__a)

							# forward pass
							with torch.no_grad():
												_UpperCamelCase									=    model(**__a)

							# verify the logits
							_UpperCamelCase									=    torch.Size((1, 10_00))
							self.assertEqual(outputs.logits.shape							,						__a)

							_UpperCamelCase									=    torch.tensor([-1.9090, -0.4993, -0.2389]).to(__a)

							self.assertTrue(torch.allclose(outputs.logits[0, :3]							,						__a							,						atol=1e-4))




		@slow
		@require_accelerate
		def 		UpperCAmelCase				(   self)			->			Any:






							'''simple docstring'''
							_UpperCamelCase									=    ViTHybridImageProcessor.from_pretrained('''google/vit-hybrid-base-bit-384''')
							_UpperCamelCase									=    ViTHybridForImageClassification.from_pretrained('''google/vit-hybrid-base-bit-384'''							,						device_map='''auto''')

							_UpperCamelCase									=    prepare_img()

							_UpperCamelCase									=    image_processor(images=__a							,						return_tensors='''pt''')
							_UpperCamelCase									=    model(**__a)
							_UpperCamelCase									=    outputs.logits
							# model predicts one of the 1000 ImageNet classes
							_UpperCamelCase									=    logits.argmax(-1).item()

							self.assertTrue(model.config.idalabel[predicted_class_idx]							,						'''tabby, tabby cat''')






