






'''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,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_DEFAULT_MEAN,
    IMAGENET_DEFAULT_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


_A     :      Optional[Any]   					=logging.get_logger(__name__)

class _lowercase     (    _lowercase		):
							a    				=	["""pixel_values"""]




							def __init__(							self:    Any      ,							UpperCamelCase__:    bool = True      ,							UpperCamelCase__:    Dict[str, int] = None      ,							UpperCamelCase__:    int = 0.9      ,							UpperCamelCase__:    PILImageResampling = PILImageResampling.BICUBIC      ,							UpperCamelCase__:    bool = True      ,							UpperCamelCase__:    Dict[str, int] = None      ,							UpperCamelCase__:    Union[int, float] = 1 / 255      ,							UpperCamelCase__:    bool = True      ,							UpperCamelCase__:    bool = True      ,							UpperCamelCase__:    Optional[Union[float, List[float]]] = None      ,							UpperCamelCase__:    Optional[Union[float, List[float]]] = None      ,							**UpperCamelCase__:    List[str]      ,							):
														super().__init__(**UpperCamelCase__					)
														lowerCamelCase__ :     Any	  =	size if size is not None else {"""shortest_edge""": 224}
														lowerCamelCase__ :     int	  =	get_size_dict(UpperCamelCase__      ,							default_to_square=UpperCamelCase__					)
														lowerCamelCase__ :     str	  =	crop_size if crop_size is not None else {"""height""": 224, """width""": 224}
														lowerCamelCase__ :     Any	  =	get_size_dict(UpperCamelCase__      ,							param_name="""crop_size"""					)

														lowerCamelCase__ :     str	  =	do_resize
														lowerCamelCase__ :     List[Any]	  =	size
														lowerCamelCase__ :     List[Any]	  =	crop_pct
														lowerCamelCase__ :     Tuple	  =	resample
														lowerCamelCase__ :     Any	  =	do_center_crop
														lowerCamelCase__ :     Tuple	  =	crop_size
														lowerCamelCase__ :     str	  =	do_rescale
														lowerCamelCase__ :     Any	  =	rescale_factor
														lowerCamelCase__ :     Union[str, Any]	  =	do_normalize
														lowerCamelCase__ :     Optional[Any]	  =	image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
														lowerCamelCase__ :     Dict	  =	image_std if image_std is not None else IMAGENET_DEFAULT_STD




							def 		lowerCamelCase_    (							self:    Any      ,							UpperCamelCase__:    np.ndarray      ,							UpperCamelCase__:    Dict[str, int]      ,							UpperCamelCase__:    Optional[float] = None      ,							UpperCamelCase__:    PILImageResampling = PILImageResampling.BICUBIC      ,							UpperCamelCase__:    Optional[Union[str, ChannelDimension]] = None      ,							**UpperCamelCase__:    Optional[int]      ,							):
														lowerCamelCase__ :     Optional[Any]	  =	get_size_dict(UpperCamelCase__      ,							default_to_square=UpperCamelCase__					)
														if "shortest_edge" not in size and ("height" not in size or "width" not in size):
																					raise ValueError(F'''size must contain \'height\' and \'width\' or \'shortest_edge\' as keys. Got {size.keys()}'''					)
														if crop_pct is not None:
																					if "shortest_edge" in size:
																												lowerCamelCase__ :     List[Any]	  =	int(size["""shortest_edge"""] / crop_pct					)
																					elif "height" in size and "width" in size:
																												if size["height"] == size["width"]:
																																			lowerCamelCase__ :     Tuple	  =	int(size["""height"""] / crop_pct					)
																												else:
																																			lowerCamelCase__ :     List[str]	  =	(int(size["""height"""] / crop_pct					), int(size["""width"""] / crop_pct					))
																					else:
																												raise ValueError("""Invalid size for resize: {}""".format(UpperCamelCase__					)					)

																					lowerCamelCase__ :     Optional[int]	  =	get_resize_output_image_size(UpperCamelCase__      ,							size=UpperCamelCase__      ,							default_to_square=UpperCamelCase__					)
														else:
																					if "shortest_edge" in size:
																												lowerCamelCase__ :     List[str]	  =	get_resize_output_image_size(UpperCamelCase__      ,							size=size["""shortest_edge"""]      ,							default_to_square=UpperCamelCase__					)
																					elif "height" in size and "width" in size:
																												lowerCamelCase__ :     Optional[Any]	  =	(size["""height"""], size["""width"""])
																					else:
																												raise ValueError("""Invalid size for resize: {}""".format(UpperCamelCase__					)					)

														return resize(UpperCamelCase__      ,							size=UpperCamelCase__      ,							resample=UpperCamelCase__      ,							data_format=UpperCamelCase__      ,							**UpperCamelCase__					)




							def 		lowerCamelCase_    (							self:    int      ,							UpperCamelCase__:    np.ndarray      ,							UpperCamelCase__:    Dict[str, int]      ,							UpperCamelCase__:    Optional[Union[str, ChannelDimension]] = None      ,							**UpperCamelCase__:    Tuple      ,							):
														lowerCamelCase__ :     int	  =	get_size_dict(UpperCamelCase__					)
														if "height" not in size or "width" not in size:
																					raise ValueError(F'''size must contain \'height\' and \'width\' as keys. Got {size.keys()}'''					)
														return center_crop(UpperCamelCase__      ,							size=(size["""height"""], size["""width"""])      ,							data_format=UpperCamelCase__      ,							**UpperCamelCase__					)




							def 		lowerCamelCase_    (							self:    Tuple      ,							UpperCamelCase__:    np.ndarray      ,							UpperCamelCase__:    Union[int, float]      ,							UpperCamelCase__:    Optional[Union[str, ChannelDimension]] = None      ,							**UpperCamelCase__:    Dict      ,							):
														return rescale(UpperCamelCase__      ,							scale=UpperCamelCase__      ,							data_format=UpperCamelCase__      ,							**UpperCamelCase__					)




							def 		lowerCamelCase_    (							self:    Any      ,							UpperCamelCase__:    np.ndarray      ,							UpperCamelCase__:    Union[float, List[float]]      ,							UpperCamelCase__:    Union[float, List[float]]      ,							UpperCamelCase__:    Optional[Union[str, ChannelDimension]] = None      ,							**UpperCamelCase__:    Optional[Any]      ,							):
														return normalize(UpperCamelCase__      ,							mean=UpperCamelCase__      ,							std=UpperCamelCase__      ,							data_format=UpperCamelCase__      ,							**UpperCamelCase__					)




							def 		lowerCamelCase_    (							self:    List[str]      ,							UpperCamelCase__:    ImageInput      ,							UpperCamelCase__:    bool = None      ,							UpperCamelCase__:    Dict[str, int] = None      ,							UpperCamelCase__:    int = None      ,							UpperCamelCase__:    PILImageResampling = None      ,							UpperCamelCase__:    bool = None      ,							UpperCamelCase__:    Dict[str, int] = None      ,							UpperCamelCase__:    bool = None      ,							UpperCamelCase__:    float = None      ,							UpperCamelCase__:    bool = None      ,							UpperCamelCase__:    Optional[Union[float, List[float]]] = None      ,							UpperCamelCase__:    Optional[Union[float, List[float]]] = None      ,							UpperCamelCase__:    Optional[Union[str, TensorType]] = None      ,							UpperCamelCase__:    ChannelDimension = ChannelDimension.FIRST      ,							**UpperCamelCase__:    Tuple      ,							):
														lowerCamelCase__ :     int	  =	do_resize if do_resize is not None else self.do_resize
														lowerCamelCase__ :     List[str]	  =	crop_pct if crop_pct is not None else self.crop_pct
														lowerCamelCase__ :     Dict	  =	resample if resample is not None else self.resample
														lowerCamelCase__ :     int	  =	do_center_crop if do_center_crop is not None else self.do_center_crop
														lowerCamelCase__ :     List[Any]	  =	do_rescale if do_rescale is not None else self.do_rescale
														lowerCamelCase__ :     Any	  =	rescale_factor if rescale_factor is not None else self.rescale_factor
														lowerCamelCase__ :     List[str]	  =	do_normalize if do_normalize is not None else self.do_normalize
														lowerCamelCase__ :     str	  =	image_mean if image_mean is not None else self.image_mean
														lowerCamelCase__ :     Dict	  =	image_std if image_std is not None else self.image_std

														lowerCamelCase__ :     Union[str, Any]	  =	size if size is not None else self.size
														lowerCamelCase__ :     Any	  =	get_size_dict(UpperCamelCase__      ,							default_to_square=UpperCamelCase__					)
														lowerCamelCase__ :     Dict	  =	crop_size if crop_size is not None else self.crop_size
														lowerCamelCase__ :     Union[str, Any]	  =	get_size_dict(UpperCamelCase__      ,							param_name="""crop_size"""					)

														lowerCamelCase__ :     Dict	  =	make_list_of_images(UpperCamelCase__					)

														if not valid_images(UpperCamelCase__					):
																					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_pct is None:
																					raise ValueError("""Crop_pct 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.
														lowerCamelCase__ :     List[Any]	  =	[to_numpy_array(UpperCamelCase__					) for image in images]

														if do_resize:
																					lowerCamelCase__ :     List[str]	  =	[self.resize(image=UpperCamelCase__      ,							size=UpperCamelCase__      ,							crop_pct=UpperCamelCase__      ,							resample=UpperCamelCase__					) for image in images]

														if do_center_crop:
																					lowerCamelCase__ :     Any	  =	[self.center_crop(image=UpperCamelCase__      ,							size=UpperCamelCase__					) for image in images]

														if do_rescale:
																					lowerCamelCase__ :     int	  =	[self.rescale(image=UpperCamelCase__      ,							scale=UpperCamelCase__					) for image in images]

														if do_normalize:
																					lowerCamelCase__ :     Optional[int]	  =	[self.normalize(image=UpperCamelCase__      ,							mean=UpperCamelCase__      ,							std=UpperCamelCase__					) for image in images]

														lowerCamelCase__ :     Tuple	  =	[to_channel_dimension_format(UpperCamelCase__      ,							UpperCamelCase__					) for image in images]

														lowerCamelCase__ :     Tuple	  =	{"""pixel_values""": images}
														return BatchFeature(data=UpperCamelCase__      ,							tensor_type=UpperCamelCase__					)









'''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     :      Optional[Any]   					=logging.get_logger(__name__)

@add_end_docstrings(_lowercase		)
class _lowercase     (    _lowercase		):




							def __init__(							self:    Union[str, Any]      ,							**UpperCamelCase__:    str					):
														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[Any]      ,							UpperCamelCase__:    Union[np.ndarray, bytes, str]      ,							**UpperCamelCase__:    int					):
														return super().__call__(UpperCamelCase__      ,							**UpperCamelCase__					)




							def 		lowerCamelCase_    (							self:    Optional[Any]      ,							**UpperCamelCase__:    int					):
														lowerCamelCase__ :     Optional[Any]	  =	{}
														if "candidate_labels" in kwargs:
																					lowerCamelCase__ :     str	  =	kwargs["""candidate_labels"""]
														if "hypothesis_template" in kwargs:
																					lowerCamelCase__ :     int	  =	kwargs["""hypothesis_template"""]

														return preprocess_params, {}, {}




							def 		lowerCamelCase_    (							self:    Optional[Any]      ,							UpperCamelCase__:    int      ,							UpperCamelCase__:    List[str]=None      ,							UpperCamelCase__:    Dict="This is a sound of {}."					):
														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
																												lowerCamelCase__ :     int	  =	requests.get(UpperCamelCase__					).content
																					else:
																												with open(UpperCamelCase__      ,							"""rb"""					) as f:
																																			lowerCamelCase__ :     Dict	  =	f.read()

														if isinstance(UpperCamelCase__      ,							UpperCamelCase__					):
																					lowerCamelCase__ :     str	  =	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"""					)

														lowerCamelCase__ :     Optional[Any]	  =	self.feature_extractor(
														    [audio]      ,							sampling_rate=self.feature_extractor.sampling_rate      ,							return_tensors="""pt"""					)
														lowerCamelCase__ :     Any	  =	candidate_labels
														lowerCamelCase__ :     Any	  =	[hypothesis_template.format(UpperCamelCase__					) for x in candidate_labels]
														lowerCamelCase__ :     int	  =	self.tokenizer(UpperCamelCase__      ,							return_tensors=self.framework      ,							padding=UpperCamelCase__					)
														lowerCamelCase__ :     Tuple	  =	[text_inputs]
														return inputs




							def 		lowerCamelCase_    (							self:    Tuple      ,							UpperCamelCase__:    Union[str, Any]					):
														lowerCamelCase__ :     Any	  =	model_inputs.pop("""candidate_labels"""					)
														lowerCamelCase__ :     Union[str, Any]	  =	model_inputs.pop("""text_inputs"""					)
														if isinstance(text_inputs[0]      ,							UpperCamelCase__					):
																					lowerCamelCase__ :     Optional[int]	  =	text_inputs[0]
														else:
																					# Batching case.
																					lowerCamelCase__ :     Tuple	  =	text_inputs[0][0]

														lowerCamelCase__ :     Tuple	  =	self.model(**UpperCamelCase__      ,							**UpperCamelCase__					)

														lowerCamelCase__ :     str	  =	{
														    """candidate_labels""": candidate_labels,
														    """logits""": outputs.logits_per_audio,
														}
														return model_outputs




							def 		lowerCamelCase_    (							self:    List[str]      ,							UpperCamelCase__:    Union[str, Any]					):
														lowerCamelCase__ :     List[str]	  =	model_outputs.pop("""candidate_labels"""					)
														lowerCamelCase__ :     int	  =	model_outputs["""logits"""][0]

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

														lowerCamelCase__ :     Tuple	  =	[
														    {"""score""": score, """label""": candidate_label}
														    for score, candidate_label in sorted(zip(UpperCamelCase__      ,							UpperCamelCase__					)      ,							key=lambda UpperCamelCase__					: -x[0]					)
														]
														return result

