




"""simple docstring"""

import json
import os
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import CLIPTokenizer, CLIPTokenizerFast
from transformers.models.clip.tokenization_clip import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
						from PIL import Image

						from transformers import CLIPImageProcessor, CLIPProcessor



@require_vision
class 			_snake_case					(				unittest.TestCase							):

				'''simple docstring'''




				def 					snake_case_			(				self   :	Optional[Any]  ):
								UpperCAmelCase_       :Optional[Any]         =   tempfile.mkdtemp()

								# fmt: off
								UpperCAmelCase_       :Optional[int]         =   ['''l''', '''o''', '''w''', '''e''', '''r''', '''s''', '''t''', '''i''', '''d''', '''n''', '''lo''', '''l</w>''', '''w</w>''', '''r</w>''', '''t</w>''', '''low</w>''', '''er</w>''', '''lowest</w>''', '''newer</w>''', '''wider''', '''<unk>''', '''<|startoftext|>''', '''<|endoftext|>''']
								# fmt: on
								UpperCAmelCase_       :int         =   dict(zip(__A       ,				range(len(__A  )  )  )  )
								UpperCAmelCase_       :List[Any]         =   ['''#version: 0.2''', '''l o''', '''lo w</w>''', '''e r</w>''', '''''']
								UpperCAmelCase_       :Optional[Any]         =   {'''unk_token''': '''<unk>'''}

								UpperCAmelCase_       :List[str]         =   os.path.join(self.tmpdirname       ,				VOCAB_FILES_NAMES['''vocab_file''']  )
								UpperCAmelCase_       :List[Any]         =   os.path.join(self.tmpdirname       ,				VOCAB_FILES_NAMES['''merges_file''']  )
								with open(self.vocab_file       ,				'''w'''       ,				encoding='''utf-8'''  ) as fp:
												fp.write(json.dumps(__A  ) + '''\n'''  )
								with open(self.merges_file       ,				'''w'''       ,				encoding='''utf-8'''  ) as fp:
												fp.write('''\n'''.join(__A  )  )

								UpperCAmelCase_       :Tuple         =   {
								    '''do_resize''': True,
								    '''size''': 20,
								    '''do_center_crop''': True,
								    '''crop_size''': 18,
								    '''do_normalize''': True,
								    '''image_mean''': [0.48_145_466, 0.4_578_275, 0.40_821_073],
								    '''image_std''': [0.26_862_954, 0.26_130_258, 0.27_577_711],
								}
								UpperCAmelCase_       :Optional[int]         =   os.path.join(self.tmpdirname       ,				__A  )
								with open(self.image_processor_file       ,				'''w'''       ,				encoding='''utf-8'''  ) as fp:
												json.dump(__A       ,				__A  )





				def 					snake_case_			(				self   :	Optional[Any]       ,				**snake_case   :	Optional[int]  ):
								return CLIPTokenizer.from_pretrained(self.tmpdirname       ,				**__A  )





				def 					snake_case_			(				self   :	Optional[int]       ,				**snake_case   :	List[str]  ):
								return CLIPTokenizerFast.from_pretrained(self.tmpdirname       ,				**__A  )





				def 					snake_case_			(				self   :	str       ,				**snake_case   :	Dict  ):
								return CLIPImageProcessor.from_pretrained(self.tmpdirname       ,				**__A  )





				def 					snake_case_			(				self   :	Union[str, Any]  ):
								shutil.rmtree(self.tmpdirname  )





				def 					snake_case_			(				self   :	str  ):
								UpperCAmelCase_       :Optional[Any]         =   [np.random.randint(255       ,				size=(3, 30, 400)       ,				dtype=np.uinta  )]

								UpperCAmelCase_       :Optional[int]         =   [Image.fromarray(np.moveaxis(__A       ,				0       ,				-1  )  ) for x in image_inputs]

								return image_inputs





				def 					snake_case_			(				self   :	Dict  ):
								UpperCAmelCase_       :Optional[int]         =   self.get_tokenizer()
								UpperCAmelCase_       :List[str]         =   self.get_rust_tokenizer()
								UpperCAmelCase_       :Union[str, Any]         =   self.get_image_processor()

								UpperCAmelCase_       :Any         =   CLIPProcessor(tokenizer=__A       ,				image_processor=__A  )
								processor_slow.save_pretrained(self.tmpdirname  )
								UpperCAmelCase_       :List[Any]         =   CLIPProcessor.from_pretrained(self.tmpdirname       ,				use_fast=__A  )

								UpperCAmelCase_       :Optional[int]         =   CLIPProcessor(tokenizer=__A       ,				image_processor=__A  )
								processor_fast.save_pretrained(self.tmpdirname  )
								UpperCAmelCase_       :int         =   CLIPProcessor.from_pretrained(self.tmpdirname  )

								self.assertEqual(processor_slow.tokenizer.get_vocab()       ,				tokenizer_slow.get_vocab()  )
								self.assertEqual(processor_fast.tokenizer.get_vocab()       ,				tokenizer_fast.get_vocab()  )
								self.assertEqual(tokenizer_slow.get_vocab()       ,				tokenizer_fast.get_vocab()  )
								self.assertIsInstance(processor_slow.tokenizer       ,				__A  )
								self.assertIsInstance(processor_fast.tokenizer       ,				__A  )

								self.assertEqual(processor_slow.image_processor.to_json_string()       ,				image_processor.to_json_string()  )
								self.assertEqual(processor_fast.image_processor.to_json_string()       ,				image_processor.to_json_string()  )
								self.assertIsInstance(processor_slow.image_processor       ,				__A  )
								self.assertIsInstance(processor_fast.image_processor       ,				__A  )





				def 					snake_case_			(				self   :	Tuple  ):
								UpperCAmelCase_       :Tuple         =   CLIPProcessor(tokenizer=self.get_tokenizer()       ,				image_processor=self.get_image_processor()  )
								processor.save_pretrained(self.tmpdirname  )

								UpperCAmelCase_       :int         =   self.get_tokenizer(bos_token='''(BOS)'''       ,				eos_token='''(EOS)'''  )
								UpperCAmelCase_       :Optional[Any]         =   self.get_image_processor(do_normalize=__A       ,				padding_value=1.0  )

								UpperCAmelCase_       :Dict         =   CLIPProcessor.from_pretrained(
								    self.tmpdirname       ,				bos_token='''(BOS)'''       ,				eos_token='''(EOS)'''       ,				do_normalize=__A       ,				padding_value=1.0  )

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

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





				def 					snake_case_			(				self   :	Any  ):
								UpperCAmelCase_       :Any         =   self.get_image_processor()
								UpperCAmelCase_       :Any         =   self.get_tokenizer()

								UpperCAmelCase_       :Optional[Any]         =   CLIPProcessor(tokenizer=__A       ,				image_processor=__A  )

								UpperCAmelCase_       :List[Any]         =   self.prepare_image_inputs()

								UpperCAmelCase_       :Optional[int]         =   image_processor(__A       ,				return_tensors='''np'''  )
								UpperCAmelCase_       :Union[str, Any]         =   processor(images=__A       ,				return_tensors='''np'''  )

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





				def 					snake_case_			(				self   :	Optional[Any]  ):
								UpperCAmelCase_       :Optional[int]         =   self.get_image_processor()
								UpperCAmelCase_       :List[str]         =   self.get_tokenizer()

								UpperCAmelCase_       :List[str]         =   CLIPProcessor(tokenizer=__A       ,				image_processor=__A  )

								UpperCAmelCase_       :Tuple         =   '''lower newer'''

								UpperCAmelCase_       :Tuple         =   processor(text=__A  )

								UpperCAmelCase_       :str         =   tokenizer(__A  )

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





				def 					snake_case_			(				self   :	Any  ):
								UpperCAmelCase_       :Dict         =   self.get_image_processor()
								UpperCAmelCase_       :Any         =   self.get_tokenizer()

								UpperCAmelCase_       :str         =   CLIPProcessor(tokenizer=__A       ,				image_processor=__A  )

								UpperCAmelCase_       :List[Any]         =   '''lower newer'''
								UpperCAmelCase_       :Optional[Any]         =   self.prepare_image_inputs()

								UpperCAmelCase_       :Tuple         =   processor(text=__A       ,				images=__A  )

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

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





				def 					snake_case_			(				self   :	List[Any]  ):
								UpperCAmelCase_       :List[str]         =   self.get_image_processor()
								UpperCAmelCase_       :int         =   self.get_tokenizer()

								UpperCAmelCase_       :int         =   CLIPProcessor(tokenizer=__A       ,				image_processor=__A  )

								UpperCAmelCase_       :List[str]         =   [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

								UpperCAmelCase_       :List[str]         =   processor.batch_decode(__A  )
								UpperCAmelCase_       :Dict         =   tokenizer.batch_decode(__A  )

								self.assertListEqual(__A       ,				__A  )





				def 					snake_case_			(				self   :	Optional[Any]  ):
								UpperCAmelCase_       :Any         =   self.get_image_processor()
								UpperCAmelCase_       :Optional[int]         =   self.get_tokenizer()

								UpperCAmelCase_       :str         =   CLIPProcessor(tokenizer=__A       ,				image_processor=__A  )

								UpperCAmelCase_       :Any         =   '''lower newer'''
								UpperCAmelCase_       :Optional[Any]         =   self.prepare_image_inputs()

								UpperCAmelCase_       :Optional[Any]         =   processor(text=__A       ,				images=__A  )

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









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