


"""simple docstring"""



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 _lowerCAmelCase				(      unittest.TestCase	):







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

       '''simple docstring'''




       snake_case    :      int      					=    0







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

       '''simple docstring'''




       snake_case    :      Optional[Any]      					=    AutoImageProcessor.from_pretrained("openai/clip-vit-base-patch32"						)
       self.assertIsInstance(lowercase_						,	lowercase_						)







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

       '''simple docstring'''




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

           snake_case    :      List[Any]      					=    AutoImageProcessor.from_pretrained(lowercase_						)
           self.assertIsInstance(lowercase_						,	lowercase_						)







   def  lowerCamelCase		(	self						)	->		Tuple:

       '''simple docstring'''




       with tempfile.TemporaryDirectory() as tmpdirname:
           snake_case    :      Optional[int]      					=    Path(lowercase_						) / "preprocessor_config.json"
           snake_case    :      Tuple      					=    Path(lowercase_						) / "config.json"
           json.dump(
               {"feature_extractor_type": "CLIPFeatureExtractor", "processor_class": "CLIPProcessor"}						,	open(lowercase_						,	"w"						)						,	)
           json.dump({"model_type": "clip"}						,	open(lowercase_						,	"w"						)						)

           snake_case    :      str      					=    AutoImageProcessor.from_pretrained(lowercase_						)
           self.assertIsInstance(lowercase_						,	lowercase_						)







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

       '''simple docstring'''




       with tempfile.TemporaryDirectory() as tmpdirname:
           snake_case    :      str      					=    CLIPConfig()

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

           # remove image_processor_type to make sure config.json alone is enough to load image processor locally
           snake_case    :      Union[str, Any]      					=    AutoImageProcessor.from_pretrained(lowercase_						).to_dict()

           config_dict.pop("image_processor_type"						)
           snake_case    :      Tuple      					=    CLIPImageProcessor(**lowercase_						)

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

           snake_case    :      List[str]      					=    AutoImageProcessor.from_pretrained(lowercase_						)

           # make sure private variable is not incorrectly saved
           snake_case    :      Any      					=    json.loads(config.to_json_string()						)
           self.assertTrue("_processor_class" not in dict_as_saved						)

       self.assertIsInstance(lowercase_						,	lowercase_						)







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

       '''simple docstring'''




       with tempfile.TemporaryDirectory() as tmpdirname:
           snake_case    :      List[Any]      					=    Path(lowercase_						) / "preprocessor_config.json"
           json.dump(
               {"image_processor_type": "CLIPImageProcessor", "processor_class": "CLIPProcessor"}						,	open(lowercase_						,	"w"						)						,	)

           snake_case    :      Dict      					=    AutoImageProcessor.from_pretrained(lowercase_						)
           self.assertIsInstance(lowercase_						,	lowercase_						)







   def  lowerCamelCase		(	self						)	->		Dict:

       '''simple docstring'''




       with self.assertRaisesRegex(
           lowercase_						,	"clip-base is not a local folder and is not a valid model identifier"						):
           snake_case    :      Optional[Any]      					=    AutoImageProcessor.from_pretrained("clip-base"						)







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

       '''simple docstring'''




       with self.assertRaisesRegex(
           lowercase_						,	r"aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"						):
           snake_case    :      Union[str, Any]      					=    AutoImageProcessor.from_pretrained(lowercase_						,	revision="aaaaaa"						)







   def  lowerCamelCase		(	self						)	->		Tuple:

       '''simple docstring'''




       with self.assertRaisesRegex(
           lowercase_						,	"hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."						,	):
           snake_case    :      Optional[Any]      					=    AutoImageProcessor.from_pretrained("hf-internal-testing/config-no-model"						)







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

       '''simple docstring'''




       with self.assertRaises(lowercase_						):
           snake_case    :      Optional[Any]      					=    AutoImageProcessor.from_pretrained("hf-internal-testing/test_dynamic_image_processor"						)
       # If remote code is disabled, we can't load this config.
       with self.assertRaises(lowercase_						):
           snake_case    :      List[Any]      					=    AutoImageProcessor.from_pretrained(
               "hf-internal-testing/test_dynamic_image_processor"						,	trust_remote_code=lowercase_						)

       snake_case    :      Optional[int]      					=    AutoImageProcessor.from_pretrained(
           "hf-internal-testing/test_dynamic_image_processor"						,	trust_remote_code=lowercase_						)
       self.assertEqual(image_processor.__class__.__name__						,	"NewImageProcessor"						)

       # Test image processor can be reloaded.
       with tempfile.TemporaryDirectory() as tmp_dir:
           image_processor.save_pretrained(lowercase_						)
           snake_case    :      str      					=    AutoImageProcessor.from_pretrained(lowercase_						,	trust_remote_code=lowercase_						)
       self.assertEqual(reloaded_image_processor.__class__.__name__						,	"NewImageProcessor"						)







   def  lowerCamelCase		(	self						)	->		int:

       '''simple docstring'''




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

           with tempfile.TemporaryDirectory() as tmpdirname:
               snake_case    :      Tuple      					=    Path(lowercase_						) / "preprocessor_config.json"
               snake_case    :      str      					=    Path(lowercase_						) / "config.json"
               json.dump(
                   {"feature_extractor_type": "CLIPFeatureExtractor", "processor_class": "CLIPProcessor"}						,	open(lowercase_						,	"w"						)						,	)
               json.dump({"model_type": "clip"}						,	open(lowercase_						,	"w"						)						)

               snake_case    :      Optional[Any]      					=    CustomImageProcessor.from_pretrained(lowercase_						)

           # 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(lowercase_						)
               snake_case    :      Optional[Any]      					=    AutoImageProcessor.from_pretrained(lowercase_						)
               self.assertIsInstance(lowercase_						,	lowercase_						)

       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  lowerCamelCase		(	self						)	->		List[str]:

       '''simple docstring'''




       class _lowerCAmelCase				(      lowerCamelCase__	):
          __UpperCAmelCase :       List[Any]							  =  True

       try:
           AutoConfig.register("custom"						,	lowercase_						)
           AutoImageProcessor.register(lowercase_						,	lowercase_						)
           # If remote code is not set, the default is to use local
           snake_case    :      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    :      Optional[int]      					=    AutoImageProcessor.from_pretrained(
               "hf-internal-testing/test_dynamic_image_processor"						,	trust_remote_code=lowercase_						)
           self.assertEqual(image_processor.__class__.__name__						,	"NewImageProcessor"						)
           self.assertTrue(image_processor.is_local						)

           # If remote is enabled, we load from the Hub
           snake_case    :      List[str]      					=    AutoImageProcessor.from_pretrained(
               "hf-internal-testing/test_dynamic_image_processor"						,	trust_remote_code=lowercase_						)
           self.assertEqual(image_processor.__class__.__name__						,	"NewImageProcessor"						)
           self.assertTrue(not hasattr(lowercase_						,	"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]








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


lowerCamelCase           =					logging.get_logger(__name__)







@add_end_docstrings(lowerCamelCase__							)
class 						__magic_name__			(							lowerCamelCase__							):

   '''simple docstring'''




   def __init__(				self,    **lowercase_       )							->    Dict:






     """simple docstring"""






     super().__init__(**lowercase_       )

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



   def __call__(				self,    lowercase_,    **lowercase_       )							->    Tuple:






     """simple docstring"""






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



   def  _UpperCAmelCase (				self,    **lowercase_       )							->    int:






     """simple docstring"""






     a__  ={}
     if "candidate_labels" in kwargs:
       a__  =kwargs['''candidate_labels''']
     if "hypothesis_template" in kwargs:
       a__  =kwargs['''hypothesis_template''']

     return preprocess_params, {}, {}



   def  _UpperCAmelCase (				self,    lowercase_,    lowercase_=None,    lowercase_="This is a sound of {}."       )							->    Union[str, Any]:






     """simple docstring"""






     if isinstance(lowercase_,    lowercase_       ):
       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
         a__  =requests.get(lowercase_       ).content
       else:
         with open(lowercase_,    '''rb'''       ) as f:
           a__  =f.read()

     if isinstance(lowercase_,    lowercase_       ):
       a__  =ffmpeg_read(lowercase_,    self.feature_extractor.sampling_rate       )

     if not isinstance(lowercase_,    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'''       )

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



   def  _UpperCAmelCase (				self,    lowercase_       )							->    str:






     """simple docstring"""






     a__  =model_inputs.pop('''candidate_labels'''       )
     a__  =model_inputs.pop('''text_inputs'''       )
     if isinstance(text_inputs[0],    lowercase_       ):
       a__  =text_inputs[0]
     else:
       # Batching case.
       a__  =text_inputs[0][0]

     a__  =self.model(**lowercase_,    **lowercase_       )

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



   def  _UpperCAmelCase (				self,    lowercase_       )							->    Any:






     """simple docstring"""






     a__  =model_outputs.pop('''candidate_labels'''       )
     a__  =model_outputs['''logits'''][0]

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

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