





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

lowerCAmelCase_        =							logging.get_logger(__name__)



@add_end_docstrings(_a	)
class    __lowerCAmelCase			(     _a	):






			def __init__(self   ,				**__magic_name__       )	->  Tuple:


							'''simple docstring'''



							super().__init__(**__magic_name__       )

							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   ,				__magic_name__   ,				**__magic_name__       )	->  Dict:


							'''simple docstring'''



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






			def      lowerCamelCase    (self   ,				**__magic_name__       )	->  List[str]:


							'''simple docstring'''



							snake_case_		:       Dict           =    {}
							if "candidate_labels" in kwargs:
											snake_case_		:       Optional[int]           =    kwargs['''candidate_labels''']
							if "hypothesis_template" in kwargs:
											snake_case_		:       Tuple           =    kwargs['''hypothesis_template''']

							return preprocess_params, {}, {}






			def      lowerCamelCase    (self   ,				__magic_name__   ,				__magic_name__=None   ,				__magic_name__="This is a photo of {}."       )	->  List[Any]:


							'''simple docstring'''



							snake_case_		:       Union[str, Any]           =    load_image(__magic_name__       )
							snake_case_		:       Optional[int]           =    self.image_processor(images=[image]   ,				return_tensors=self.framework       )
							snake_case_		:       str           =    candidate_labels
							snake_case_		:       Any           =    [hypothesis_template.format(__magic_name__       ) for x in candidate_labels]
							snake_case_		:       List[Any]           =    self.tokenizer(__magic_name__   ,				return_tensors=self.framework   ,				padding=__magic_name__       )
							snake_case_		:       str           =    [text_inputs]
							return inputs






			def      lowerCamelCase    (self   ,				__magic_name__       )	->  List[Any]:


							'''simple docstring'''



							snake_case_		:       Dict           =    model_inputs.pop('''candidate_labels'''       )
							snake_case_		:       Optional[Any]           =    model_inputs.pop('''text_inputs'''       )
							if isinstance(text_inputs[0]   ,				__magic_name__       ):
											snake_case_		:       Dict           =    text_inputs[0]
							else:
											# Batching case.
											snake_case_		:       List[str]           =    text_inputs[0][0]

							snake_case_		:       str           =    self.model(**__magic_name__   ,				**__magic_name__       )

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






			def      lowerCamelCase    (self   ,				__magic_name__       )	->  str:


							'''simple docstring'''



							snake_case_		:       Optional[int]           =    model_outputs.pop('''candidate_labels'''       )
							snake_case_		:       Optional[int]           =    model_outputs['''logits'''][0]
							if self.framework == "pt":
											snake_case_		:       Dict           =    logits.softmax(dim=-1       ).squeeze(-1       )
											snake_case_		:       Optional[Any]           =    probs.tolist()
											if not isinstance(__magic_name__   ,				__magic_name__       ):
															snake_case_		:       Optional[Any]           =    [scores]
							elif self.framework == "tf":
											snake_case_		:       Tuple           =    stable_softmax(__magic_name__   ,				axis=-1       )
											snake_case_		:       List[str]           =    probs.numpy().tolist()
							else:
											raise ValueError(F'''Unsupported framework: {self.framework}'''       )

							snake_case_		:       Union[str, Any]           =    [
							    {'''score''': score, '''label''': candidate_label}
							    for score, candidate_label in sorted(zip(__magic_name__   ,				__magic_name__       )   ,				key=lambda __magic_name__       : -x[0]       )
							]
							return result




import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


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

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowerCAmelCase__    :Optional[Any]       =				get_tests_dir('''fixtures''')

class 						__a (							unittest.TestCase     ):
		def       UpperCAmelCase__							(     self		)					->			Any:




									"""simple docstring"""





									_UpperCAmelCase											=     mock.Mock()
									_UpperCAmelCase											=     500
									_UpperCAmelCase											=     {}
									_UpperCAmelCase											=     HTTPError
									_UpperCAmelCase											=     {}

									# Download this model to make sure it's in the cache.
									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'		)
									# Under the mock environment we get a 500 error when trying to reach the model.
									with mock.patch('requests.Session.request'     ,       return_value=_SCREAMING_SNAKE_CASE		) as mock_head:
																_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'		)
																# This check we did call the fake head request
																mock_head.assert_called()






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




									"""simple docstring"""





									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(
									    'https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'		)



@is_staging_test
class 						__a (							unittest.TestCase     ):
		@classmethod
		def       UpperCAmelCase__							(     cls		)					->			Tuple:




									"""simple docstring"""





									_UpperCAmelCase											=     TOKEN
									HfFolder.save_token(_SCREAMING_SNAKE_CASE		)
		@classmethod
		def       UpperCAmelCase__							(     cls		)					->			Tuple:




									"""simple docstring"""





									try:
																delete_repo(token=cls._token     ,       repo_id='test-feature-extractor'		)
									except HTTPError:
																pass

									try:
																delete_repo(token=cls._token     ,       repo_id='valid_org/test-feature-extractor-org'		)
									except HTTPError:
																pass

									try:
																delete_repo(token=cls._token     ,       repo_id='test-dynamic-feature-extractor'		)
									except HTTPError:
																pass
		def       UpperCAmelCase__							(     self		)					->			Optional[int]:




									"""simple docstring"""





									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(_SCREAMING_SNAKE_CASE		)
									feature_extractor.push_to_hub('test-feature-extractor'     ,       use_auth_token=self._token		)

									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor'''		)
									for k, v in feature_extractor.__dict__.items():
																self.assertEqual(_SCREAMING_SNAKE_CASE     ,       getattr(_SCREAMING_SNAKE_CASE     ,       _SCREAMING_SNAKE_CASE		)		)

									# Reset repo
									delete_repo(token=self._token     ,       repo_id='test-feature-extractor'		)

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
																feature_extractor.save_pretrained(
																    _SCREAMING_SNAKE_CASE     ,       repo_id='test-feature-extractor'     ,       push_to_hub=_SCREAMING_SNAKE_CASE     ,       use_auth_token=self._token		)

									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor'''		)
									for k, v in feature_extractor.__dict__.items():
																self.assertEqual(_SCREAMING_SNAKE_CASE     ,       getattr(_SCREAMING_SNAKE_CASE     ,       _SCREAMING_SNAKE_CASE		)		)
		def       UpperCAmelCase__							(     self		)					->			Optional[Any]:




									"""simple docstring"""





									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(_SCREAMING_SNAKE_CASE		)
									feature_extractor.push_to_hub('valid_org/test-feature-extractor'     ,       use_auth_token=self._token		)

									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor'		)
									for k, v in feature_extractor.__dict__.items():
																self.assertEqual(_SCREAMING_SNAKE_CASE     ,       getattr(_SCREAMING_SNAKE_CASE     ,       _SCREAMING_SNAKE_CASE		)		)

									# Reset repo
									delete_repo(token=self._token     ,       repo_id='valid_org/test-feature-extractor'		)

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
																feature_extractor.save_pretrained(
																    _SCREAMING_SNAKE_CASE     ,       repo_id='valid_org/test-feature-extractor-org'     ,       push_to_hub=_SCREAMING_SNAKE_CASE     ,       use_auth_token=self._token		)

									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor-org'		)
									for k, v in feature_extractor.__dict__.items():
																self.assertEqual(_SCREAMING_SNAKE_CASE     ,       getattr(_SCREAMING_SNAKE_CASE     ,       _SCREAMING_SNAKE_CASE		)		)






		def       UpperCAmelCase__							(     self		)					->			Tuple:




									"""simple docstring"""





									CustomFeatureExtractor.register_for_auto_class()
									_UpperCAmelCase											=     CustomFeatureExtractor.from_pretrained(_SCREAMING_SNAKE_CASE		)

									feature_extractor.push_to_hub('test-dynamic-feature-extractor'     ,       use_auth_token=self._token		)

									# This has added the proper auto_map field to the config
									self.assertDictEqual(
									    feature_extractor.auto_map     ,       {'AutoFeatureExtractor': 'custom_feature_extraction.CustomFeatureExtractor'}     ,       )

									_UpperCAmelCase											=     AutoFeatureExtractor.from_pretrained(
									    f'''{USER}/test-dynamic-feature-extractor'''     ,       trust_remote_code=_SCREAMING_SNAKE_CASE		)
									# Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
									self.assertEqual(new_feature_extractor.__class__.__name__     ,       'CustomFeatureExtractor'		)
