






'''simple docstring'''





from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import center_crop, normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
  import PIL


SCREAMING_SNAKE_CASE 							=						logging.get_logger(__name__)
class 							A_    (						UpperCamelCase__			):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:			Tuple =					["pixel_values"]


 def __init__(	self							,  _A = True							,  _A = None							,  _A = PIL.Image.BICUBIC							,  _A = True							,  _A = None							,  _A = 1 / 255							,  _A = True							,  _A = True							,  _A = None							,  _A = None							,  **_A							,  ) ->		None:

     """simple docstring"""



     super().__init__(**_a)
     _UpperCAmelCase				:      Union[str, Any]			  =				size if size is not None else {"""height""": 256, """width""": 256}
     _UpperCAmelCase				:      List[Any]			  =				get_size_dict(_a)
     _UpperCAmelCase				:      Tuple			  =				crop_size if crop_size is not None else {"""height""": 224, """width""": 224}
     _UpperCAmelCase				:      Union[str, Any]			  =				get_size_dict(_a							,  param_name='''crop_size''')

     _UpperCAmelCase				:      str			  =				do_resize
     _UpperCAmelCase				:      Optional[int]			  =				size
     _UpperCAmelCase				:      Any			  =				resample
     _UpperCAmelCase				:      Any			  =				do_center_crop
     _UpperCAmelCase				:      List[Any]			  =				crop_size
     _UpperCAmelCase				:      Any			  =				do_rescale
     _UpperCAmelCase				:      List[Any]			  =				rescale_factor
     _UpperCAmelCase				:      Union[str, Any]			  =				do_normalize
     _UpperCAmelCase				:      Tuple			  =				image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
     _UpperCAmelCase				:      int			  =				image_std if image_std is not None else IMAGENET_STANDARD_STD


 def  snake_case__				(	self							,  _A							,  _A							,  _A = PIL.Image.BICUBIC							,  _A = None							,  **_A							,  ) ->		np.ndarray:

     """simple docstring"""



     _UpperCAmelCase				:      List[Any]			  =				get_size_dict(_a)
     if "height" not in size or "width" not in size:
         raise ValueError(f'''The size dictionary must have keys \'height\' and \'width\'. Got {size.keys()}''')
     return resize(
         _a							,  size=(size['''height'''], size['''width'''])							,  resample=_a							,  data_format=_a							,  **_a)


 def  snake_case__				(	self							,  _A							,  _A							,  _A = None							,  **_A							,  ) ->		np.ndarray:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[Any]			  =				get_size_dict(_a)
     if "height" not in size or "width" not in size:
         raise ValueError(f'''The size dictionary must have keys \'height\' and \'width\'. Got {size.keys()}''')
     return center_crop(_a							,  size=(size['''height'''], size['''width'''])							,  data_format=_a							,  **_a)


 def  snake_case__				(	self							,  _A							,  _A							,  _A = None							,  **_A							,  ) ->		Optional[int]:

     """simple docstring"""



     return rescale(_a							,  scale=_a							,  data_format=_a							,  **_a)


 def  snake_case__				(	self							,  _A							,  _A							,  _A							,  _A = None							,  **_A							,  ) ->		np.ndarray:

     """simple docstring"""



     return normalize(_a							,  mean=_a							,  std=_a							,  data_format=_a							,  **_a)


 def  snake_case__				(	self							,  _A							,  _A = None							,  _A = None							,  _A=None							,  _A = None							,  _A = None							,  _A = None							,  _A = None							,  _A = None							,  _A = None							,  _A = None							,  _A = None							,  _A = ChannelDimension.FIRST							,  **_A							,  ) ->		PIL.Image.Image:

     """simple docstring"""



     _UpperCAmelCase				:      int			  =				do_resize if do_resize is not None else self.do_resize
     _UpperCAmelCase				:      List[str]			  =				resample if resample is not None else self.resample
     _UpperCAmelCase				:      Optional[int]			  =				do_center_crop if do_center_crop is not None else self.do_center_crop
     _UpperCAmelCase				:      Any			  =				do_rescale if do_rescale is not None else self.do_rescale
     _UpperCAmelCase				:      str			  =				rescale_factor if rescale_factor is not None else self.rescale_factor
     _UpperCAmelCase				:      Tuple			  =				do_normalize if do_normalize is not None else self.do_normalize
     _UpperCAmelCase				:      Any			  =				image_mean if image_mean is not None else self.image_mean
     _UpperCAmelCase				:      List[Any]			  =				image_std if image_std is not None else self.image_std

     _UpperCAmelCase				:      List[Any]			  =				size if size is not None else self.size
     _UpperCAmelCase				:      Optional[int]			  =				get_size_dict(_a)
     _UpperCAmelCase				:      List[str]			  =				crop_size if crop_size is not None else self.crop_size
     _UpperCAmelCase				:      Any			  =				get_size_dict(_a							,  param_name='''crop_size''')

     _UpperCAmelCase				:      Tuple			  =				make_list_of_images(_a)

     if not valid_images(_a):
         raise ValueError(
             '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
             '''torch.Tensor, tf.Tensor or jax.ndarray.''')

     if do_resize and size is None or resample is None:
         raise ValueError('''Size and resample must be specified if do_resize is True.''')

     if do_center_crop and crop_size is None:
         raise ValueError('''Crop size must be specified if do_center_crop is True.''')

     if do_rescale and rescale_factor is None:
         raise ValueError('''Rescale factor must be specified if do_rescale is True.''')

     if do_normalize and (image_mean is None or image_std is None):
         raise ValueError('''Image mean and std must be specified if do_normalize is True.''')

     # All transformations expect numpy arrays.
     _UpperCAmelCase				:      Dict			  =				[to_numpy_array(_a) for image in images]

     if do_resize:
         _UpperCAmelCase				:      Union[str, Any]			  =				[self.resize(image=_a							,  size=_a							,  resample=_a) for image in images]

     if do_center_crop:
         _UpperCAmelCase				:      Tuple			  =				[self.center_crop(image=_a							,  size=_a) for image in images]

     if do_rescale:
         _UpperCAmelCase				:      Any			  =				[self.rescale(image=_a							,  scale=_a) for image in images]

     if do_normalize:
         _UpperCAmelCase				:      Union[str, Any]			  =				[self.normalize(image=_a							,  mean=_a							,  std=_a) for image in images]

     _UpperCAmelCase				:      Optional[int]			  =				[to_channel_dimension_format(_a							,  _a) for image in images]

     _UpperCAmelCase				:      int			  =				{"""pixel_values""": images}
     return BatchFeature(data=_a							,  tensor_type=_a)












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


SCREAMING_SNAKE_CASE 							=						logging.get_logger(__name__)
@add_end_docstrings(__lowercase			)
class 							A_    (						__lowercase			):






 '''simple docstring'''




 def __init__(	self							,  **_A) ->		int:

     """simple docstring"""



     super().__init__(**_A)

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


 def __call__(	self							,  _A							,  **_A) ->		List[str]:

     """simple docstring"""



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


 def  snake_case__				(	self							,  **_A) ->		List[str]:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				{}
     if "candidate_labels" in kwargs:
         _UpperCAmelCase				:      Union[str, Any]			  =				kwargs['''candidate_labels''']
     if "hypothesis_template" in kwargs:
         _UpperCAmelCase				:      Dict			  =				kwargs['''hypothesis_template''']

     return preprocess_params, {}, {}


 def  snake_case__				(	self							,  _A							,  _A=None							,  _A="This is a sound of {}.") ->		Tuple:

     """simple docstring"""



     if isinstance(_A							,  _A):
         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				:      List[Any]			  =				requests.get(_A).content
         else:
             with open(_A							,  '''rb''') as f:
                 _UpperCAmelCase				:      List[str]			  =				f.read()

     if isinstance(_A							,  _A):
         _UpperCAmelCase				:      List[Any]			  =				ffmpeg_read(_A							,  self.feature_extractor.sampling_rate)

     if not isinstance(_A							,  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				:      Optional[Any]			  =				self.feature_extractor(
         [audio]							,  sampling_rate=self.feature_extractor.sampling_rate							,  return_tensors='''pt''')
     _UpperCAmelCase				:      Dict			  =				candidate_labels
     _UpperCAmelCase				:      List[str]			  =				[hypothesis_template.format(_A) for x in candidate_labels]
     _UpperCAmelCase				:      Optional[Any]			  =				self.tokenizer(_A							,  return_tensors=self.framework							,  padding=_A)
     _UpperCAmelCase				:      str			  =				[text_inputs]
     return inputs


 def  snake_case__				(	self							,  _A) ->		List[str]:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				model_inputs.pop('''candidate_labels''')
     _UpperCAmelCase				:      Union[str, Any]			  =				model_inputs.pop('''text_inputs''')
     if isinstance(text_inputs[0]							,  _A):
         _UpperCAmelCase				:      int			  =				text_inputs[0]
     else:
         # Batching case.
         _UpperCAmelCase				:      Dict			  =				text_inputs[0][0]

     _UpperCAmelCase				:      str			  =				self.model(**_A							,  **_A)

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


 def  snake_case__				(	self							,  _A) ->		List[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      str			  =				model_outputs.pop('''candidate_labels''')
     _UpperCAmelCase				:      str			  =				model_outputs['''logits'''][0]

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

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




