





import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    AutoConfig,
    AutoImageProcessor,
    CLIPConfig,
    CLIPImageProcessor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER


sys.path.append(str(Path(__file__).parent.parent.parent.parent / """utils"""))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_image_processing import CustomImageProcessor  # noqa E402
class SCREAMING_SNAKE_CASE_  (    unittest.TestCase    ):






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







							"""simple docstring"""


							snake_case_						:	Optional[int]      =		0






						def UpperCAmelCase_ (					self					:					Optional[Any]		)							->			Union[str, Any]:







							"""simple docstring"""


							snake_case_						:	int      =		AutoImageProcessor.from_pretrained('openai/clip-vit-base-patch32'		)
							self.assertIsInstance(_A						,      _A		)






						def UpperCAmelCase_ (					self					:					Union[str, Any]		)							->			List[Any]:







							"""simple docstring"""


							with tempfile.TemporaryDirectory() as tmpdirname:
								snake_case_						:	Dict      =		Path(_A		) / '''preprocessor_config.json'''
								snake_case_						:	Optional[Any]      =		Path(_A		) / '''config.json'''
								json.dump(
								    {'image_processor_type': 'CLIPImageProcessor', 'processor_class': 'CLIPProcessor'}						,      open(_A						,      'w'		)						,      )
								json.dump({'model_type': 'clip'}						,      open(_A						,      'w'		)		)

								snake_case_						:	Optional[Any]      =		AutoImageProcessor.from_pretrained(_A		)
								self.assertIsInstance(_A						,      _A		)






						def UpperCAmelCase_ (					self					:					Optional[int]		)							->			str:







							"""simple docstring"""


							with tempfile.TemporaryDirectory() as tmpdirname:
								snake_case_						:	int      =		Path(_A		) / '''preprocessor_config.json'''
								snake_case_						:	Tuple      =		Path(_A		) / '''config.json'''
								json.dump(
								    {'feature_extractor_type': 'CLIPFeatureExtractor', 'processor_class': 'CLIPProcessor'}						,      open(_A						,      'w'		)						,      )
								json.dump({'model_type': 'clip'}						,      open(_A						,      'w'		)		)

								snake_case_						:	List[str]      =		AutoImageProcessor.from_pretrained(_A		)
								self.assertIsInstance(_A						,      _A		)






						def UpperCAmelCase_ (					self					:					Union[str, Any]		)							->			List[str]:







							"""simple docstring"""


							with tempfile.TemporaryDirectory() as tmpdirname:
								snake_case_						:	List[Any]      =		CLIPConfig()

								# Create a dummy config file with image_proceesor_type
								snake_case_						:	Any      =		Path(_A		) / '''preprocessor_config.json'''
								snake_case_						:	List[Any]      =		Path(_A		) / '''config.json'''
								json.dump(
								    {'image_processor_type': 'CLIPImageProcessor', 'processor_class': 'CLIPProcessor'}						,      open(_A						,      'w'		)						,      )
								json.dump({'model_type': 'clip'}						,      open(_A						,      'w'		)		)

								# remove image_processor_type to make sure config.json alone is enough to load image processor locally
								snake_case_						:	List[str]      =		AutoImageProcessor.from_pretrained(_A		).to_dict()

								config_dict.pop('image_processor_type'		)
								snake_case_						:	List[str]      =		CLIPImageProcessor(**_A		)

								# save in new folder
								model_config.save_pretrained(_A		)
								config.save_pretrained(_A		)

								snake_case_						:	Optional[int]      =		AutoImageProcessor.from_pretrained(_A		)

								# make sure private variable is not incorrectly saved
								snake_case_						:	int      =		json.loads(config.to_json_string()		)
								self.assertTrue('_processor_class' not in dict_as_saved		)

							self.assertIsInstance(_A						,      _A		)






						def UpperCAmelCase_ (					self					:					List[Any]		)							->			Tuple:







							"""simple docstring"""


							with tempfile.TemporaryDirectory() as tmpdirname:
								snake_case_						:	Optional[Any]      =		Path(_A		) / '''preprocessor_config.json'''
								json.dump(
								    {'image_processor_type': 'CLIPImageProcessor', 'processor_class': 'CLIPProcessor'}						,      open(_A						,      'w'		)						,      )

								snake_case_						:	Tuple      =		AutoImageProcessor.from_pretrained(_A		)
								self.assertIsInstance(_A						,      _A		)






						def UpperCAmelCase_ (					self					:					Any		)							->			Tuple:







							"""simple docstring"""


							with self.assertRaisesRegex(
							    _A						,      'clip-base is not a local folder and is not a valid model identifier'		):
								snake_case_						:	Union[str, Any]      =		AutoImageProcessor.from_pretrained('clip-base'		)






						def UpperCAmelCase_ (					self					:					str		)							->			Any:







							"""simple docstring"""


							with self.assertRaisesRegex(
							    _A						,      R'aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'		):
								snake_case_						:	str      =		AutoImageProcessor.from_pretrained(_A						,      revision='aaaaaa'		)






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







							"""simple docstring"""


							with self.assertRaisesRegex(
							    _A						,      'hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json.'						,      ):
								snake_case_						:	Tuple      =		AutoImageProcessor.from_pretrained('hf-internal-testing/config-no-model'		)






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







							"""simple docstring"""


							with self.assertRaises(_A		):
								snake_case_						:	Dict      =		AutoImageProcessor.from_pretrained('hf-internal-testing/test_dynamic_image_processor'		)
							# If remote code is disabled, we can't load this config.
							with self.assertRaises(_A		):
								snake_case_						:	Union[str, Any]      =		AutoImageProcessor.from_pretrained(
								    'hf-internal-testing/test_dynamic_image_processor'						,      trust_remote_code=_A		)

							snake_case_						:	List[Any]      =		AutoImageProcessor.from_pretrained(
							    'hf-internal-testing/test_dynamic_image_processor'						,      trust_remote_code=_A		)
							self.assertEqual(image_processor.__class__.__name__						,      'NewImageProcessor'		)

							# Test image processor can be reloaded.
							with tempfile.TemporaryDirectory() as tmp_dir:
								image_processor.save_pretrained(_A		)
								snake_case_						:	Union[str, Any]      =		AutoImageProcessor.from_pretrained(_A						,      trust_remote_code=_A		)
							self.assertEqual(reloaded_image_processor.__class__.__name__						,      'NewImageProcessor'		)






						def UpperCAmelCase_ (					self					:					List[str]		)							->			Tuple:







							"""simple docstring"""


							try:
								AutoConfig.register('custom'						,      _A		)
								AutoImageProcessor.register(_A						,      _A		)
								# Trying to register something existing in the Transformers library will raise an error
								with self.assertRaises(_A		):
									AutoImageProcessor.register(_A						,      _A		)

								with tempfile.TemporaryDirectory() as tmpdirname:
									snake_case_						:	Any      =		Path(_A		) / '''preprocessor_config.json'''
									snake_case_						:	Any      =		Path(_A		) / '''config.json'''
									json.dump(
									    {'feature_extractor_type': 'CLIPFeatureExtractor', 'processor_class': 'CLIPProcessor'}						,      open(_A						,      'w'		)						,      )
									json.dump({'model_type': 'clip'}						,      open(_A						,      'w'		)		)

									snake_case_						:	Any      =		CustomImageProcessor.from_pretrained(_A		)

								# Now that the config is registered, it can be used as any other config with the auto-API
								with tempfile.TemporaryDirectory() as tmp_dir:
									image_processor.save_pretrained(_A		)
									snake_case_						:	List[str]      =		AutoImageProcessor.from_pretrained(_A		)
									self.assertIsInstance(_A						,      _A		)

							finally:
								if "custom" in CONFIG_MAPPING._extra_content:
									del CONFIG_MAPPING._extra_content["custom"]
								if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
									del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]






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







							"""simple docstring"""


							class SCREAMING_SNAKE_CASE_  (    __snake_case    ):
													__magic_name__:	Optional[int]       					=				True

							try:
								AutoConfig.register('custom'						,      _A		)
								AutoImageProcessor.register(_A						,      _A		)
								# If remote code is not set, the default is to use local
								snake_case_						:	List[Any]      =		AutoImageProcessor.from_pretrained('hf-internal-testing/test_dynamic_image_processor'		)
								self.assertEqual(image_processor.__class__.__name__						,      'NewImageProcessor'		)
								self.assertTrue(image_processor.is_local		)

								# If remote code is disabled, we load the local one.
								snake_case_						:	Union[str, Any]      =		AutoImageProcessor.from_pretrained(
								    'hf-internal-testing/test_dynamic_image_processor'						,      trust_remote_code=_A		)
								self.assertEqual(image_processor.__class__.__name__						,      'NewImageProcessor'		)
								self.assertTrue(image_processor.is_local		)

								# If remote is enabled, we load from the Hub
								snake_case_						:	int      =		AutoImageProcessor.from_pretrained(
								    'hf-internal-testing/test_dynamic_image_processor'						,      trust_remote_code=_A		)
								self.assertEqual(image_processor.__class__.__name__						,      'NewImageProcessor'		)
								self.assertTrue(not hasattr(_A						,      'is_local'		)		)

							finally:
								if "custom" in CONFIG_MAPPING._extra_content:
									del CONFIG_MAPPING._extra_content["custom"]
								if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
									del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]



'''simple docstring'''




from typing import Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging
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_VISUAL_QUESTION_ANSWERING_MAPPING

lowercase			:				str    		=			logging.get_logger(__name__)





@add_end_docstrings(__snake_case						)
class  A				(							__snake_case						):





     def __init__(							self						,			*SCREAMING_SNAKE_CASE						,			**SCREAMING_SNAKE_CASE	) ->				List[Any]:






         """simple docstring"""






         super().__init__(*SCREAMING_SNAKE_CASE						,			**SCREAMING_SNAKE_CASE	)
         self.check_model_type(SCREAMING_SNAKE_CASE	)





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE=None						,			SCREAMING_SNAKE_CASE=None						,			SCREAMING_SNAKE_CASE=None						,			**SCREAMING_SNAKE_CASE	) ->				Tuple:






         """simple docstring"""






         A,  A       :   Dict					=						{}, {}
         if padding is not None:
             A       :   List[str]					=						padding
         if truncation is not None:
             A       :   Dict					=						truncation
         if top_k is not None:
             A       :   Optional[Any]					=						top_k
         return preprocess_params, {}, postprocess_params





     def __call__(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE = None						,			**SCREAMING_SNAKE_CASE	) ->				Dict:






         """simple docstring"""






         if isinstance(SCREAMING_SNAKE_CASE						,			(Image.Image, str)	) and isinstance(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	):
             A       :   int					=						{'''image''': image, '''question''': question}
         else:
             A       :   Any					=						image
         A       :   Any					=						super().__call__(SCREAMING_SNAKE_CASE						,			**SCREAMING_SNAKE_CASE	)
         return results





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE=False						,			SCREAMING_SNAKE_CASE=False	) ->				Any:






         """simple docstring"""






         A       :   Union[str, Any]					=						load_image(inputs['''image''']	)
         A       :   Optional[Any]					=						self.tokenizer(
             inputs['''question''']						,			return_tensors=self.framework						,			padding=SCREAMING_SNAKE_CASE						,			truncation=SCREAMING_SNAKE_CASE	)
         A       :   Dict					=						self.image_processor(images=SCREAMING_SNAKE_CASE						,			return_tensors=self.framework	)
         model_inputs.update(SCREAMING_SNAKE_CASE	)
         return model_inputs





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE	) ->				List[str]:






         """simple docstring"""






         A       :   List[Any]					=						self.model(**SCREAMING_SNAKE_CASE	)
         return model_outputs





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE=5	) ->				int:






         """simple docstring"""






         if top_k > self.model.config.num_labels:
             A       :   Dict					=						self.model.config.num_labels

         if self.framework == "pt":
             A       :   Optional[int]					=						model_outputs.logits.sigmoid()[0]
             A,  A       :   int					=						probs.topk(SCREAMING_SNAKE_CASE	)
         else:
             raise ValueError(F'Unsupported framework: {self.framework}'	)

         A       :   int					=						scores.tolist()
         A       :   List[str]					=						ids.tolist()
         return [{"score": score, "answer": self.model.config.idalabel[_id]} for score, _id in zip(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)]




