



"""simple docstring"""


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 AutoImageProcessor, ViTImageProcessor
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_image_processing import CustomImageProcessor  # noqa E402


__A   :							Optional[int]										=   get_tests_dir("fixtures")
class 			_a (					unittest.TestCase):
			"""simple docstring"""



			def  lowercase__						(		self     :    List[str]   )->Dict:
							# A mock response for an HTTP head request to emulate server down
							_UpperCAmelCase					=    mock.Mock()
							_UpperCAmelCase					=    5_0_0
							_UpperCAmelCase					=    {}
							_UpperCAmelCase					=    HTTPError
							_UpperCAmelCase					=    {}

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



			def  lowercase__						(		self     :    List[Any]   )->int:
							# This test is for deprecated behavior and can be removed in v5
							_UpperCAmelCase					=    ViTImageProcessor.from_pretrained(
							    '''https://huggingface.co/hf-internal-testing/tiny-random-vit/resolve/main/preprocessor_config.json'''   )



			def  lowercase__						(		self     :    Any   )->Tuple:
							with self.assertRaises(__UpperCamelCase   ):
											# config is in subfolder, the following should not work without specifying the subfolder
											_UpperCAmelCase					=    AutoImageProcessor.from_pretrained('''hf-internal-testing/stable-diffusion-all-variants'''   )

							_UpperCAmelCase					=    AutoImageProcessor.from_pretrained(
							    '''hf-internal-testing/stable-diffusion-all-variants'''				,   subfolder='''feature_extractor'''   )

							self.assertIsNotNone(__UpperCamelCase   )



@is_staging_test
class 			_a (					unittest.TestCase):
			"""simple docstring"""



			@classmethod
			def  lowercase__						(		cls     :    Dict   )->List[Any]:
							_UpperCAmelCase					=    TOKEN
							HfFolder.save_token(__UpperCamelCase   )



			@classmethod
			def  lowercase__						(		cls     :    int   )->List[str]:
							try:
											delete_repo(token=cls._token				,   repo_id='''test-image-processor'''   )
							except HTTPError:
											pass

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

							try:
											delete_repo(token=cls._token				,   repo_id='''test-dynamic-image-processor'''   )
							except HTTPError:
											pass



			def  lowercase__						(		self     :    Tuple   )->List[str]:
							_UpperCAmelCase					=    ViTImageProcessor.from_pretrained(__UpperCamelCase   )
							image_processor.push_to_hub('''test-image-processor'''				,   use_auth_token=self._token   )

							_UpperCAmelCase					=    ViTImageProcessor.from_pretrained(F'{USER}/test-image-processor'   )
							for k, v in image_processor.__dict__.items():
											self.assertEqual(__UpperCamelCase				,   getattr(__UpperCamelCase				,   __UpperCamelCase   )   )

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

							# Push to hub via save_pretrained
							with tempfile.TemporaryDirectory() as tmp_dir:
											image_processor.save_pretrained(
											    __UpperCamelCase				,   repo_id='''test-image-processor'''				,   push_to_hub=__UpperCamelCase				,   use_auth_token=self._token   )

							_UpperCAmelCase					=    ViTImageProcessor.from_pretrained(F'{USER}/test-image-processor'   )
							for k, v in image_processor.__dict__.items():
											self.assertEqual(__UpperCamelCase				,   getattr(__UpperCamelCase				,   __UpperCamelCase   )   )



			def  lowercase__						(		self     :    Union[str, Any]   )->int:
							_UpperCAmelCase					=    ViTImageProcessor.from_pretrained(__UpperCamelCase   )
							image_processor.push_to_hub('''valid_org/test-image-processor'''				,   use_auth_token=self._token   )

							_UpperCAmelCase					=    ViTImageProcessor.from_pretrained('''valid_org/test-image-processor'''   )
							for k, v in image_processor.__dict__.items():
											self.assertEqual(__UpperCamelCase				,   getattr(__UpperCamelCase				,   __UpperCamelCase   )   )

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

							# Push to hub via save_pretrained
							with tempfile.TemporaryDirectory() as tmp_dir:
											image_processor.save_pretrained(
											    __UpperCamelCase				,   repo_id='''valid_org/test-image-processor-org'''				,   push_to_hub=__UpperCamelCase				,   use_auth_token=self._token   )

							_UpperCAmelCase					=    ViTImageProcessor.from_pretrained('''valid_org/test-image-processor-org'''   )
							for k, v in image_processor.__dict__.items():
											self.assertEqual(__UpperCamelCase				,   getattr(__UpperCamelCase				,   __UpperCamelCase   )   )



			def  lowercase__						(		self     :    Dict   )->str:
							CustomImageProcessor.register_for_auto_class()
							_UpperCAmelCase					=    CustomImageProcessor.from_pretrained(__UpperCamelCase   )

							image_processor.push_to_hub('''test-dynamic-image-processor'''				,   use_auth_token=self._token   )

							# This has added the proper auto_map field to the config
							self.assertDictEqual(
							    image_processor.auto_map				,   {'''AutoImageProcessor''': '''custom_image_processing.CustomImageProcessor'''}				,   )

							_UpperCAmelCase					=    AutoImageProcessor.from_pretrained(
							    F'{USER}/test-dynamic-image-processor'				,   trust_remote_code=__UpperCamelCase   )
							# Can't make an isinstance check because the new_image_processor is from the CustomImageProcessor class of a dynamic module
							self.assertEqual(new_image_processor.__class__.__name__				,   '''CustomImageProcessor'''   )








"""simple docstring"""


from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


__A   :							List[str]										=   logging.get_logger(__name__)
@add_end_docstrings(lowerCAmelCase)
class 			_a (					lowerCAmelCase):
			"""simple docstring"""



			def __init__(		self     :    Union[str, Any]				,   **__UpperCamelCase     :    Optional[Any]   )->List[Any]:
							super().__init__(**__UpperCamelCase   )

							if self.framework != "pt":
											raise ValueError(F'The {self.__class__} is only available in PyTorch.'   )
							# No specific FOR_XXX available yet



			def __call__(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[np.ndarray, bytes, str]				,   **__UpperCamelCase     :    Tuple   )->List[str]:
							return super().__call__(__UpperCamelCase				,   **__UpperCamelCase   )



			def  lowercase__						(		self     :    Any				,   **__UpperCamelCase     :    Any   )->Union[str, Any]:
							_UpperCAmelCase					=    {}
							if "candidate_labels" in kwargs:
											_UpperCAmelCase					=    kwargs['''candidate_labels''']
							if "hypothesis_template" in kwargs:
											_UpperCAmelCase					=    kwargs['''hypothesis_template''']

							return preprocess_params, {}, {}



			def  lowercase__						(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[str, Any]				,   __UpperCamelCase     :    List[Any]=None				,   __UpperCamelCase     :    Tuple="This is a sound of {}."   )->int:
							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											if audio.startswith('''http://'''   ) or audio.startswith('''https://'''   ):
															# We need to actually check for a real protocol, otherwise it's impossible to use a local file
															# like http_huggingface_co.png
															_UpperCAmelCase					=    requests.get(__UpperCamelCase   ).content
											else:
															with open(__UpperCamelCase				,   '''rb'''   ) as f:
																			_UpperCAmelCase					=    f.read()

							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    ffmpeg_read(__UpperCamelCase				,   self.feature_extractor.sampling_rate   )

							if not isinstance(__UpperCamelCase				,   np.ndarray   ):
											raise ValueError('''We expect a numpy ndarray as input'''   )
							if len(audio.shape   ) != 1:
											raise ValueError('''We expect a single channel audio input for ZeroShotAudioClassificationPipeline'''   )

							_UpperCAmelCase					=    self.feature_extractor(
							    [audio]				,   sampling_rate=self.feature_extractor.sampling_rate				,   return_tensors='''pt'''   )
							_UpperCAmelCase					=    candidate_labels
							_UpperCAmelCase					=    [hypothesis_template.format(__UpperCamelCase   ) for x in candidate_labels]
							_UpperCAmelCase					=    self.tokenizer(__UpperCamelCase				,   return_tensors=self.framework				,   padding=__UpperCamelCase   )
							_UpperCAmelCase					=    [text_inputs]
							return inputs



			def  lowercase__						(		self     :    Union[str, Any]				,   __UpperCamelCase     :    Optional[int]   )->Any:
							_UpperCAmelCase					=    model_inputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_inputs.pop('''text_inputs'''   )
							if isinstance(text_inputs[0]				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    text_inputs[0]
							else:
											# Batching case.
											_UpperCAmelCase					=    text_inputs[0][0]

							_UpperCAmelCase					=    self.model(**__UpperCamelCase				,   **__UpperCamelCase   )

							_UpperCAmelCase					=    {
							    '''candidate_labels''': candidate_labels,
							    '''logits''': outputs.logits_per_audio,
							}
							return model_outputs



			def  lowercase__						(		self     :    List[str]				,   __UpperCamelCase     :    Any   )->List[Any]:
							_UpperCAmelCase					=    model_outputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_outputs['''logits'''][0]

							if self.framework == "pt":
											_UpperCAmelCase					=    logits.softmax(dim=0   )
											_UpperCAmelCase					=    probs.tolist()
							else:
											raise ValueError('''`tf` framework not supported.'''   )

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



