






import json
import os
import shutil
import tempfile
import unittest

import numpy as np

from transformers import BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES, BertTokenizer
from transformers.testing_utils import require_tokenizers, require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
							from PIL import Image

							from transformers import VisionTextDualEncoderProcessor, ViTImageProcessor

@require_tokenizers
@require_vision
class 				lowerCAmelCase_	(		unittest.TestCase			):


				def 						__a					(					self      ):
											_lowercase							:   Any 							=   tempfile.mkdtemp()

											# fmt: off
											_lowercase							:   Tuple 							=   ['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'want', '##want', '##ed', 'wa', 'un', 'runn', '##ing', ',', 'low', 'lowest']
											# fmt: on
											_lowercase							:   List[Any] 							=   os.path.join(self.tmpdirname ,						VOCAB_FILES_NAMES['vocab_file']      )
											with open(self.vocab_file ,						'w' ,						encoding='utf-8'      ) as vocab_writer:
																		vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]      )      )

											_lowercase							:   Dict 							=   {
											    'do_resize': True,
											    'size': {'height': 1_8, 'width': 1_8},
											    'do_normalize': True,
											    'image_mean': [0.5, 0.5, 0.5],
											    'image_std': [0.5, 0.5, 0.5],
											}
											_lowercase							:   List[Any] 							=   os.path.join(self.tmpdirname ,						snake_case__      )
											with open(self.image_processor_file ,						'w' ,						encoding='utf-8'      ) as fp:
																		json.dump(snake_case__ ,						snake_case__      )


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


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


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


				def 						__a					(					self      ):
											_lowercase							:   List[str] 							=   [np.random.randint(2_5_5 ,						size=(3, 3_0, 4_0_0) ,						dtype=np.uinta      )]

											_lowercase							:   List[str] 							=   [Image.fromarray(np.moveaxis(snake_case__ ,						0 ,						-1      )      ) for x in image_inputs]

											return image_inputs


				def 						__a					(					self      ):
											_lowercase							:   str 							=   self.get_tokenizer()
											_lowercase							:   Optional[int] 							=   self.get_image_processor()

											_lowercase							:   Tuple 							=   VisionTextDualEncoderProcessor(tokenizer=snake_case__ ,						image_processor=snake_case__      )

											processor.save_pretrained(self.tmpdirname      )
											_lowercase							:   Tuple 							=   VisionTextDualEncoderProcessor.from_pretrained(self.tmpdirname      )

											self.assertEqual(processor.tokenizer.get_vocab() ,						tokenizer.get_vocab()      )
											self.assertIsInstance(processor.tokenizer ,						(BertTokenizer, BertTokenizerFast)      )

											self.assertEqual(processor.image_processor.to_json_string() ,						image_processor.to_json_string()      )
											self.assertIsInstance(processor.image_processor ,						snake_case__      )


				def 						__a					(					self      ):
											_lowercase							:   Optional[int] 							=   VisionTextDualEncoderProcessor(
											    tokenizer=self.get_tokenizer() ,						image_processor=self.get_image_processor()      )
											processor.save_pretrained(self.tmpdirname      )

											_lowercase							:   Any 							=   self.get_tokenizer(bos_token='(BOS)' ,						eos_token='(EOS)'      )
											_lowercase							:   int 							=   self.get_image_processor(do_normalize=snake_case__ ,						padding_value=1.0      )

											_lowercase							:   Dict 							=   VisionTextDualEncoderProcessor.from_pretrained(
											    self.tmpdirname ,						bos_token='(BOS)' ,						eos_token='(EOS)' ,						do_normalize=snake_case__ ,						padding_value=1.0      )

											self.assertEqual(processor.tokenizer.get_vocab() ,						tokenizer_add_kwargs.get_vocab()      )
											self.assertIsInstance(processor.tokenizer ,						(BertTokenizer, BertTokenizerFast)      )

											self.assertEqual(processor.image_processor.to_json_string() ,						image_processor_add_kwargs.to_json_string()      )
											self.assertIsInstance(processor.image_processor ,						snake_case__      )


				def 						__a					(					self      ):
											_lowercase							:   int 							=   self.get_image_processor()
											_lowercase							:   int 							=   self.get_tokenizer()

											_lowercase							:   Tuple 							=   VisionTextDualEncoderProcessor(tokenizer=snake_case__ ,						image_processor=snake_case__      )

											_lowercase							:   Any 							=   self.prepare_image_inputs()

											_lowercase							:   str 							=   image_processor(snake_case__ ,						return_tensors='np'      )
											_lowercase							:   str 							=   processor(images=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							:   int 							=   self.get_image_processor()
											_lowercase							:   Union[str, Any] 							=   self.get_tokenizer()

											_lowercase							:   Optional[Any] 							=   VisionTextDualEncoderProcessor(tokenizer=snake_case__ ,						image_processor=snake_case__      )

											_lowercase							:   Tuple 							=   'lower newer'

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

											_lowercase							:   Any 							=   tokenizer(snake_case__      )

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


				def 						__a					(					self      ):
											_lowercase							:   Union[str, Any] 							=   self.get_image_processor()
											_lowercase							:   str 							=   self.get_tokenizer()

											_lowercase							:   List[str] 							=   VisionTextDualEncoderProcessor(tokenizer=snake_case__ ,						image_processor=snake_case__      )

											_lowercase							:   Dict 							=   'lower newer'
											_lowercase							:   List[str] 							=   self.prepare_image_inputs()

											_lowercase							:   List[str] 							=   processor(text=snake_case__ ,						images=snake_case__      )

											self.assertListEqual(list(inputs.keys()      ) ,						['input_ids', 'token_type_ids', 'attention_mask', 'pixel_values']      )

											# test if it raises when no input is passed
											with self.assertRaises(snake_case__      ):
																		processor()


				def 						__a					(					self      ):
											_lowercase							:   Dict 							=   self.get_image_processor()
											_lowercase							:   str 							=   self.get_tokenizer()

											_lowercase							:   str 							=   VisionTextDualEncoderProcessor(tokenizer=snake_case__ ,						image_processor=snake_case__      )

											_lowercase							:   List[Any] 							=   [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

											_lowercase							:   Optional[int] 							=   processor.batch_decode(snake_case__      )
											_lowercase							:   Optional[int] 							=   tokenizer.batch_decode(snake_case__      )

											self.assertListEqual(snake_case__ ,						snake_case__      )


				def 						__a					(					self      ):
											_lowercase							:   Dict 							=   self.get_image_processor()
											_lowercase							:   List[Any] 							=   self.get_tokenizer()

											_lowercase							:   Optional[int] 							=   VisionTextDualEncoderProcessor(tokenizer=snake_case__ ,						image_processor=snake_case__      )

											_lowercase							:   Optional[int] 							=   'lower newer'
											_lowercase							:   int 							=   self.prepare_image_inputs()

											_lowercase							:   Optional[Any] 							=   processor(text=snake_case__ ,						images=snake_case__      )

											self.assertListEqual(list(inputs.keys()      ) ,						processor.model_input_names      )










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