






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


				def 						__a					(					self      ):
											_lowercase							:   Union[str, Any] 							=   '| <pad> <unk> <s> </s> a b c d e f g h i j k'.split()
											_lowercase							:   int 							=   dict(zip(snake_case__ ,						range(len(snake_case__      )      )      )      )

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

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


				def 						__a					(					self ,						**_lowerCAmelCase      ):
											_lowercase							:   str 							=   self.add_kwargs_tokens_map.copy()
											kwargs.update(snake_case__      )
											return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname ,						**snake_case__      )


				def 						__a					(					self ,						**_lowerCAmelCase      ):
											return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname ,						**snake_case__      )


				def 						__a					(					self ,						**_lowerCAmelCase      ):
											return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name ,						**snake_case__      )


				def 						__a					(					self      ):
											shutil.rmtree(self.tmpdirname      )


				def 						__a					(					self      ):
											_lowercase							:   Any 							=   self.get_tokenizer()
											_lowercase							:   Optional[Any] 							=   self.get_feature_extractor()
											_lowercase							:   Any 							=   self.get_decoder()

											_lowercase							:   List[str] 							=   WavaVecaProcessorWithLM(tokenizer=snake_case__ ,						feature_extractor=snake_case__ ,						decoder=snake_case__      )

											processor.save_pretrained(self.tmpdirname      )
											_lowercase							:   Tuple 							=   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 						__a					(					self      ):
											_lowercase							:   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
											_lowercase							:   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 						__a					(					self      ):
											_lowercase							:   str 							=   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 						__a					(					self      ):
											_lowercase							:   Tuple 							=   self.get_feature_extractor()
											_lowercase							:   Dict 							=   self.get_tokenizer()
											_lowercase							:   Tuple 							=   self.get_decoder()

											_lowercase							:   Optional[int] 							=   WavaVecaProcessorWithLM(tokenizer=snake_case__ ,						feature_extractor=snake_case__ ,						decoder=snake_case__      )

											_lowercase							:   Optional[Any] 							=   floats_list((3, 1_0_0_0)      )

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

											_lowercase							:   List[str] 							=   WavaVecaProcessorWithLM(tokenizer=snake_case__ ,						feature_extractor=snake_case__ ,						decoder=snake_case__      )

											_lowercase							:   str 							=   'This is a test string'

											_lowercase							:   List[Any] 							=   processor(text=snake_case__      )

											_lowercase							:   List[str] 							=   tokenizer(snake_case__      )

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


				def 						__a					(					self ,						_lowerCAmelCase=(2, 1_0, 1_6) ,						_lowerCAmelCase=7_7      ):
											np.random.seed(snake_case__      )
											return np.random.rand(*snake_case__      )


				def 						__a					(					self      ):
											_lowercase							:   int 							=   self.get_feature_extractor()
											_lowercase							:   Tuple 							=   self.get_tokenizer()
											_lowercase							:   List[str] 							=   self.get_decoder()

											_lowercase							:   Any 							=   WavaVecaProcessorWithLM(tokenizer=snake_case__ ,						feature_extractor=snake_case__ ,						decoder=snake_case__      )

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

											_lowercase							:   str 							=   processor.decode(snake_case__      )

											_lowercase							:   Optional[Any] 							=   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 						__a					(					self ,						_lowerCAmelCase      ):
											_lowercase							:   Any 							=   self.get_feature_extractor()
											_lowercase							:   Optional[int] 							=   self.get_tokenizer()
											_lowercase							:   List[Any] 							=   self.get_decoder()

											_lowercase							:   int 							=   WavaVecaProcessorWithLM(tokenizer=snake_case__ ,						feature_extractor=snake_case__ ,						decoder=snake_case__      )

											_lowercase							:   str 							=   self._get_dummy_logits()

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

											_lowercase							:   Optional[Any] 							=   list(snake_case__      )

											with get_context('fork'      ).Pool() as p:
																		_lowercase							:   Optional[int] 							=   decoder.decode_beams_batch(snake_case__ ,						snake_case__      )

											_lowercase							,							_lowercase							,							_lowercase							:   Dict 							=   [], [], []
											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 						__a					(					self      ):
											_lowercase							:   Any 							=   self.get_feature_extractor()
											_lowercase							:   str 							=   self.get_tokenizer()
											_lowercase							:   Tuple 							=   self.get_decoder()

											_lowercase							:   Union[str, Any] 							=   WavaVecaProcessorWithLM(tokenizer=snake_case__ ,						feature_extractor=snake_case__ ,						decoder=snake_case__      )

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

											_lowercase							:   Any 							=   1_5
											_lowercase							:   Optional[Any] 							=   -20.0
											_lowercase							:   str 							=   -4.0

											_lowercase							:   str 							=   processor.batch_decode(
											    snake_case__ ,						beam_width=snake_case__ ,						beam_prune_logp=snake_case__ ,						token_min_logp=snake_case__ ,						)
											_lowercase							:   List[str] 							=   decoded_processor_out.text

											_lowercase							:   Optional[Any] 							=   list(snake_case__      )

											with get_context('fork'      ).Pool() as pool:
																		_lowercase							:   List[Any] 							=   decoder.decode_beams_batch(
																		    snake_case__ ,						snake_case__ ,						beam_width=snake_case__ ,						beam_prune_logp=snake_case__ ,						token_min_logp=snake_case__ ,						)

											_lowercase							:   int 							=   [d[0][0] for d in decoded_decoder_out]
											_lowercase							:   Any 							=   [d[0][2] for d in decoded_decoder_out]
											_lowercase							:   Tuple 							=   [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.0_54, -18.4_47] ,						snake_case__ ,						atol=1E-3      )      )

											self.assertTrue(np.array_equal(snake_case__ ,						decoded_processor_out.lm_score      )      )
											self.assertTrue(np.allclose([-15.5_54, -13.94_74] ,						snake_case__ ,						atol=1E-3      )      )


				def 						__a					(					self      ):
											_lowercase							:   Tuple 							=   self.get_feature_extractor()
											_lowercase							:   Optional[int] 							=   self.get_tokenizer()
											_lowercase							:   List[str] 							=   self.get_decoder()

											_lowercase							:   List[str] 							=   WavaVecaProcessorWithLM(tokenizer=snake_case__ ,						feature_extractor=snake_case__ ,						decoder=snake_case__      )

											_lowercase							:   str 							=   self._get_dummy_logits()

											_lowercase							:   int 							=   2.0
											_lowercase							:   Union[str, Any] 							=   5.0
											_lowercase							:   Any 							=   -20.0
											_lowercase							:   List[str] 							=   True

											_lowercase							:   Tuple 							=   processor.batch_decode(
											    snake_case__ ,						alpha=snake_case__ ,						beta=snake_case__ ,						unk_score_offset=snake_case__ ,						lm_score_boundary=snake_case__ ,						)
											_lowercase							:   Optional[Any] 							=   decoded_processor_out.text

											_lowercase							:   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:
																		_lowercase							:   Union[str, Any] 							=   decoder.decode_beams_batch(
																		    snake_case__ ,						snake_case__ ,						)

											_lowercase							:   Tuple 							=   [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__      )
											_lowercase							:   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 						__a					(					self      ):
											_lowercase							:   Tuple 							=   WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'      )

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

											_lowercase							:   Any 							=   os.listdir(snake_case__      )
											_lowercase							:   Optional[int] 							=   ['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 						__a					(					self      ):
											_lowercase							:   str 							=   snapshot_download('hf-internal-testing/processor_with_lm'      )

											_lowercase							:   List[Any] 							=   WavaVecaProcessorWithLM.from_pretrained(snake_case__      )

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

											_lowercase							:   Dict 							=   os.listdir(snake_case__      )
											_lowercase							:   Optional[Any] 							=   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 						__a					(					self      ):
											_lowercase							:   Dict 							=   WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'      )
											_lowercase							:   Any 							=   AutoProcessor.from_pretrained('hf-internal-testing/processor_with_lm'      )

											_lowercase							:   Optional[int] 							=   floats_list((3, 1_0_0_0)      )

											_lowercase							:   Optional[int] 							=   processor_wavaveca(snake_case__ ,						return_tensors='np'      )
											_lowercase							:   Dict 							=   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      )

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

											_lowercase							:   str 							=   processor_wavaveca.batch_decode(snake_case__      )
											_lowercase							:   Optional[Any] 							=   processor_auto.batch_decode(snake_case__      )

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


				def 						__a					(					self      ):
											_lowercase							:   List[Any] 							=   self.get_feature_extractor()
											_lowercase							:   int 							=   self.get_tokenizer()
											_lowercase							:   Optional[Any] 							=   self.get_decoder()

											_lowercase							:   int 							=   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 						__a					(					_lowerCAmelCase ,						_lowerCAmelCase      ):
											_lowercase							:   List[str] 							=   [d[key] for d in offsets]
											return retrieved_list


				def 						__a					(					self      ):
											_lowercase							:   int 							=   WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'      )
											_lowercase							:   Optional[int] 							=   self._get_dummy_logits()[0]

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

											_lowercase							:   Tuple 							=   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 						__a					(					self      ):
											import torch

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

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

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

											_lowercase							:   List[str] 							=   processor.decode(logits[0] ,						output_word_offsets=snake_case__      )

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

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

											# output times
											_lowercase							:   Any 							=   torch.tensor(self.get_from_offsets(snake_case__ ,						'start_time'      )      )
											_lowercase							:   Dict 							=   torch.tensor(self.get_from_offsets(snake_case__ ,						'end_time'      )      )

											# fmt: off
											_lowercase							:   List[str] 							=   torch.tensor([1.41_99, 1.65_99, 2.25_99, 3.0, 3.24, 3.59_99, 3.79_99, 4.09_99, 4.26, 4.94, 5.28, 5.65_99, 5.78, 5.94, 6.32, 6.53_99, 6.65_99]      )
											_lowercase							:   List[str] 							=   torch.tensor([1.53_99, 1.89_99, 2.9, 3.16, 3.53_99, 3.72, 4.01_99, 4.17_99, 4.76, 5.15_99, 5.55_99, 5.69_99, 5.86, 6.19_99, 6.38, 6.61_99, 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      )      )










from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
						from PIL import Image

						from ..image_utils import load_image

if is_torch_available():
						from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
						from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
						from ..tf_utils import stable_softmax

SCREAMING_SNAKE_CASE:     Optional[int]									= logging.get_logger(__name__)



@add_end_docstrings(SCREAMING_SNAKE_CASE__ )
class    lowercase_		(SCREAMING_SNAKE_CASE__ ):



						def __init__(			self	:   Any					,							**snake_case__	:   str  ):

								"""simple docstring"""

								super().__init__(**snake_case__  )

								requires_backends(self					,							'vision'  )
								self.check_model_type(
								    TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
								    if self.framework == 'tf'
								    else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING  )



						def __call__(			self	:   str					,							snake_case__	:   Union[str, List[str], "Image", List["Image"]]					,							**snake_case__	:   Optional[Any]  ):

								"""simple docstring"""

								return super().__call__(snake_case__					,							**snake_case__  )



						def 	__a					(			self	:   str					,							**snake_case__	:   str  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							{}
								if "candidate_labels" in kwargs:
										SCREAMING_SNAKE_CASE_					      =							kwargs['candidate_labels']
								if "hypothesis_template" in kwargs:
										SCREAMING_SNAKE_CASE_					      =							kwargs['hypothesis_template']

								return preprocess_params, {}, {}



						def 	__a					(			self	:   List[str]					,							snake_case__	:   Optional[int]					,							snake_case__	:   Optional[Any]=None					,							snake_case__	:   List[str]="This is a photo of {}."  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							load_image(snake_case__  )
								SCREAMING_SNAKE_CASE_					      =							self.image_processor(images=[image]					,							return_tensors=self.framework  )
								SCREAMING_SNAKE_CASE_					      =							candidate_labels
								SCREAMING_SNAKE_CASE_					      =							[hypothesis_template.format(snake_case__  ) for x in candidate_labels]
								SCREAMING_SNAKE_CASE_					      =							self.tokenizer(snake_case__					,							return_tensors=self.framework					,							padding=snake_case__  )
								SCREAMING_SNAKE_CASE_					      =							[text_inputs]
								return inputs



						def 	__a					(			self	:   List[Any]					,							snake_case__	:   Any  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							model_inputs.pop('candidate_labels'  )
								SCREAMING_SNAKE_CASE_					      =							model_inputs.pop('text_inputs'  )
								if isinstance(text_inputs[0]					,							snake_case__  ):
										SCREAMING_SNAKE_CASE_					      =							text_inputs[0]
								else:
										# Batching case.
										SCREAMING_SNAKE_CASE_					      =							text_inputs[0][0]

								SCREAMING_SNAKE_CASE_					      =							self.model(**snake_case__					,							**snake_case__  )

								SCREAMING_SNAKE_CASE_					      =							{
								    'candidate_labels': candidate_labels,
								    'logits': outputs.logits_per_image,
								}
								return model_outputs






						def 	__a					(			self	:   List[Any]					,							snake_case__	:   Optional[Any]  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							model_outputs.pop('candidate_labels'  )
								SCREAMING_SNAKE_CASE_					      =							model_outputs['logits'][0]
								if self.framework == "pt":
										SCREAMING_SNAKE_CASE_					      =							logits.softmax(dim=-1  ).squeeze(-1  )
										SCREAMING_SNAKE_CASE_					      =							probs.tolist()
										if not isinstance(snake_case__					,							snake_case__  ):
												SCREAMING_SNAKE_CASE_					      =							[scores]
								elif self.framework == "tf":
										SCREAMING_SNAKE_CASE_					      =							stable_softmax(snake_case__					,							axis=-1  )
										SCREAMING_SNAKE_CASE_					      =							probs.numpy().tolist()
								else:
										raise ValueError(f'''Unsupported framework: {self.framework}'''  )

								SCREAMING_SNAKE_CASE_					      =							[
								    {'score': score, 'label': candidate_label}
								    for score, candidate_label in sorted(zip(snake_case__					,							snake_case__  )					,							key=lambda snake_case__  : -x[0]  )
								]
								return result