


'''simple docstring'''







import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
		from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

_SCREAMING_SNAKE_CASE       :   List[Any]        = logging.get_logger(__name__)



def 						_UpperCamelCase							( UpperCamelCase__     ,   UpperCamelCase__		):






				"""simple docstring"""



				__magic_name__				:					str						       =       F"""{sampling_rate}"""
				__magic_name__				:					Optional[int]						       =       "1"
				__magic_name__				:					Dict						       =       "f32le"
				__magic_name__				:					Dict						       =       [
				    "ffmpeg",
				    "-i",
				    "pipe:0",
				    "-ac",
				    ac,
				    "-ar",
				    ar,
				    "-f",
				    format_for_conversion,
				    "-hide_banner",
				    "-loglevel",
				    "quiet",
				    "pipe:1",
				]

				try:
								__magic_name__				:					List[str]						       =       subprocess.Popen(UpperCamelCase__     ,   stdin=subprocess.PIPE     ,   stdout=subprocess.PIPE		)
				except FileNotFoundError:
								raise ValueError("ffmpeg was not found but is required to load audio files from filename"		)
				__magic_name__				:					int						       =       ffmpeg_process.communicate(UpperCamelCase__		)
				__magic_name__				:					Any						       =       output_stream[0]

				__magic_name__				:					int						       =       np.frombuffer(UpperCamelCase__     ,   np.floataa		)
				if audio.shape[0] == 0:
								raise ValueError("Malformed soundfile"		)
				return audio






@add_end_docstrings(snake_case_		)
class 		lowerCamelCase    (				snake_case_		):






						'''simple docstring'''







						def __init__(  self:	Any      ,							*__UpperCamelCase:	List[str]      ,							**__UpperCamelCase:	Any     )							->	Union[str, Any]:
										# Default, might be overriden by the model.config.
										__magic_name__				:					Tuple						       =       5
										super().__init__(*a_      ,							**a_     )

										if self.framework != "pt":
														raise ValueError(f"""The {self.__class__} is only available in PyTorch."""     )

										self.check_model_type(a_     )







						def __call__(  self:	List[Any]      ,							__UpperCamelCase:	int      ,							**__UpperCamelCase:	List[str]      ,							)							->	int:
										return super().__call__(a_      ,							**a_     )







						def lowerCAmelCase__ (  self:	Union[str, Any]      ,							__UpperCamelCase:	Any=None      ,							**__UpperCamelCase:	Dict     )							->	Dict:
										# No parameters on this pipeline right now
										__magic_name__				:					str						       =       {}
										if top_k is not None:
														if top_k > self.model.config.num_labels:
																		__magic_name__				:					Optional[int]						       =       self.model.config.num_labels
														__magic_name__				:					Optional[Any]						       =       top_k
										return {}, {}, postprocess_params







						def lowerCAmelCase__ (  self:	Tuple      ,							__UpperCamelCase:	List[str]     )							->	str:
										if isinstance(a_      ,							a_     ):
														if inputs.startswith("http://"     ) or inputs.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
																		__magic_name__				:					Union[str, Any]						       =       requests.get(a_     ).content
														else:
																		with open(a_      ,							"rb"     ) as f:
																						__magic_name__				:					List[Any]						       =       f.read()

										if isinstance(a_      ,							a_     ):
														__magic_name__				:					Optional[int]						       =       ffmpeg_read(a_      ,							self.feature_extractor.sampling_rate     )

										if isinstance(a_      ,							a_     ):
														# Accepting `"array"` which is the key defined in `datasets` for
														# better integration
														if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
																		raise ValueError(
																		    "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
																		    "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
																		    "containing the sampling_rate associated with that array"     )

														__magic_name__				:					Optional[int]						       =       inputs.pop("raw"      ,							a_     )
														if _inputs is None:
																		# Remove path which will not be used from `datasets`.
																		inputs.pop("path"      ,							a_     )
																		__magic_name__				:					List[Any]						       =       inputs.pop("array"      ,							a_     )
														__magic_name__				:					Union[str, Any]						       =       inputs.pop("sampling_rate"     )
														__magic_name__				:					List[Any]						       =       _inputs
														if in_sampling_rate != self.feature_extractor.sampling_rate:
																		import torch

																		if is_torchaudio_available():
																						from torchaudio import functional as F
																		else:
																						raise ImportError(
																						    "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
																						    "The torchaudio package can be installed through: `pip install torchaudio`."     )

																		__magic_name__				:					Optional[Any]						       =       F.resample(
																		    torch.from_numpy(a_     )      ,							a_      ,							self.feature_extractor.sampling_rate     ).numpy()

										if not isinstance(a_      ,							np.ndarray     ):
														raise ValueError("We expect a numpy ndarray as input"     )
										if len(inputs.shape     ) != 1:
														raise ValueError("We expect a single channel audio input for AudioClassificationPipeline"     )

										__magic_name__				:					int						       =       self.feature_extractor(
										    a_      ,							sampling_rate=self.feature_extractor.sampling_rate      ,							return_tensors="pt"     )
										return processed







						def lowerCAmelCase__ (  self:	str      ,							__UpperCamelCase:	Optional[Any]     )							->	List[str]:
										__magic_name__				:					Tuple						       =       self.model(**a_     )
										return model_outputs







						def lowerCAmelCase__ (  self:	List[str]      ,							__UpperCamelCase:	Tuple      ,							__UpperCamelCase:	List[Any]=5     )							->	int:
										__magic_name__				:					int						       =       model_outputs.logits[0].softmax(-1     )
										__magic_name__       ,     __magic_name__				:					Any						       =       probs.topk(a_     )

										__magic_name__				:					Optional[Any]						       =       scores.tolist()
										__magic_name__				:					Dict						       =       ids.tolist()

										__magic_name__				:					List[str]						       =       [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(a_      ,							a_     )]

										return labels



'''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_DEFAULT_MEAN,
    IMAGENET_DEFAULT_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    get_image_size,
    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       :   int        = logging.get_logger(__name__)






class 		_snake_case    (				snake_case_		):






						'''simple docstring'''







						__snake_case   =      ["pixel_values"]







						def __init__(  self:	Union[str, Any]      ,							__UpperCamelCase:	bool = True      ,							__UpperCamelCase:	Dict[str, int] = None      ,							__UpperCamelCase:	bool = True      ,							__UpperCamelCase:	Dict[str, int] = None      ,							__UpperCamelCase:	PILImageResampling = PILImageResampling.BICUBIC      ,							__UpperCamelCase:	bool = True      ,							__UpperCamelCase:	Union[int, float] = 1 / 255      ,							__UpperCamelCase:	bool = True      ,							__UpperCamelCase:	Optional[Union[float, List[float]]] = None      ,							__UpperCamelCase:	Optional[Union[float, List[float]]] = None      ,							**__UpperCamelCase:	Union[str, Any]      ,							)							->	None:
										super().__init__(**__UpperCamelCase     )
										__magic_name__				:					Optional[Any]						       =       crop_size if crop_size is not None else {"height": 256, "width": 256}
										__magic_name__				:					Optional[int]						       =       get_size_dict(__UpperCamelCase      ,							param_name="crop_size"     )
										__magic_name__				:					Union[str, Any]						       =       size if size is not None else {"height": 224, "width": 224}
										__magic_name__				:					Tuple						       =       get_size_dict(__UpperCamelCase     )

										__magic_name__				:					Tuple						       =       do_center_crop
										__magic_name__				:					int						       =       crop_size
										__magic_name__				:					Dict						       =       do_resize
										__magic_name__				:					Optional[Any]						       =       size
										__magic_name__				:					Optional[Any]						       =       resample
										__magic_name__				:					str						       =       do_rescale
										__magic_name__				:					int						       =       rescale_factor
										__magic_name__				:					Dict						       =       do_normalize
										__magic_name__				:					Any						       =       image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
										__magic_name__				:					str						       =       image_std if image_std is not None else IMAGENET_DEFAULT_STD







						def lowerCAmelCase__ (  self:	str      ,							__UpperCamelCase:	np.ndarray      ,							__UpperCamelCase:	Dict[str, int]      ,							__UpperCamelCase:	Optional[int] = None      ,							__UpperCamelCase:	Optional[Union[str, ChannelDimension]] = None      ,							**__UpperCamelCase:	List[str]      ,							)							->	np.ndarray:
										__magic_name__				:					List[str]						       =       self.size if size is None else size
										__magic_name__				:					Dict						       =       get_size_dict(__UpperCamelCase     )
										__magic_name__				:					Dict						       =       get_size_dict(__UpperCamelCase      ,							param_name="crop_size"     )

										__magic_name__       ,     __magic_name__				:					Dict						       =       get_image_size(__UpperCamelCase     )
										__magic_name__				:					int						       =       min(__UpperCamelCase      ,							__UpperCamelCase     )
										__magic_name__				:					Dict						       =       (size["height"] / crop_size["height"]) * min_dim
										__magic_name__				:					str						       =       (size["width"] / crop_size["width"]) * min_dim
										return center_crop(__UpperCamelCase      ,							size=(cropped_height, cropped_width)      ,							data_format=__UpperCamelCase      ,							**__UpperCamelCase     )







						def lowerCAmelCase__ (  self:	int      ,							__UpperCamelCase:	np.ndarray      ,							__UpperCamelCase:	Dict[str, int]      ,							__UpperCamelCase:	PILImageResampling = PIL.Image.BICUBIC      ,							__UpperCamelCase:	Optional[Union[str, ChannelDimension]] = None      ,							**__UpperCamelCase:	Any      ,							)							->	np.ndarray:
										__magic_name__				:					Optional[Any]						       =       get_size_dict(__UpperCamelCase     )
										if "height" not in size or "width" not in size:
														raise ValueError(f"""The size dictionary must contain the keys 'height' and 'width'. Got {size.keys()}"""     )
										return resize(
										    __UpperCamelCase      ,							size=(size["height"], size["width"])      ,							resample=__UpperCamelCase      ,							data_format=__UpperCamelCase      ,							**__UpperCamelCase     )







						def lowerCAmelCase__ (  self:	Optional[int]      ,							__UpperCamelCase:	np.ndarray      ,							__UpperCamelCase:	Union[int, float]      ,							__UpperCamelCase:	Optional[Union[str, ChannelDimension]] = None      ,							**__UpperCamelCase:	List[Any]      ,							)							->	List[Any]:
										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:	str      ,							)							->	np.ndarray:
										return normalize(__UpperCamelCase      ,							mean=__UpperCamelCase      ,							std=__UpperCamelCase      ,							data_format=__UpperCamelCase      ,							**__UpperCamelCase     )







						def lowerCAmelCase__ (  self:	str      ,							__UpperCamelCase:	ImageInput      ,							__UpperCamelCase:	Optional[bool] = None      ,							__UpperCamelCase:	Optional[Dict[str, int]] = None      ,							__UpperCamelCase:	Optional[bool] = None      ,							__UpperCamelCase:	Optional[Dict[str, int]] = None      ,							__UpperCamelCase:	PILImageResampling = None      ,							__UpperCamelCase:	Optional[bool] = None      ,							__UpperCamelCase:	Optional[float] = None      ,							__UpperCamelCase:	Optional[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:	Dict      ,							)							->	PIL.Image.Image:
										__magic_name__				:					Optional[Any]						       =       do_center_crop if do_center_crop is not None else self.do_center_crop
										__magic_name__				:					Optional[Any]						       =       crop_size if crop_size is not None else self.crop_size
										__magic_name__				:					Tuple						       =       get_size_dict(__UpperCamelCase      ,							param_name="crop_size"     )
										__magic_name__				:					int						       =       do_resize if do_resize is not None else self.do_resize
										__magic_name__				:					Optional[Any]						       =       size if size is not None else self.size
										__magic_name__				:					Any						       =       get_size_dict(__UpperCamelCase     )
										__magic_name__				:					Optional[int]						       =       resample if resample is not None else self.resample
										__magic_name__				:					int						       =       do_rescale if do_rescale is not None else self.do_rescale
										__magic_name__				:					int						       =       rescale_factor if rescale_factor is not None else self.rescale_factor
										__magic_name__				:					Any						       =       do_normalize if do_normalize is not None else self.do_normalize
										__magic_name__				:					Dict						       =       image_mean if image_mean is not None else self.image_mean
										__magic_name__				:					Tuple						       =       image_std if image_std is not None else self.image_std

										__magic_name__				:					List[Any]						       =       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_center_crop and crop_size is None:
														raise ValueError("If `do_center_crop` is set to `True`, `crop_size` must be provided."     )

										if do_resize and size is None:
														raise ValueError("Size must be specified if do_resize 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 image standard deviation must be specified if do_normalize is True."     )

										# All transformations expect numpy arrays.
										__magic_name__				:					Optional[Any]						       =       [to_numpy_array(__UpperCamelCase     ) for image in images]

										if do_center_crop:
														__magic_name__				:					Union[str, Any]						       =       [self.center_crop(__UpperCamelCase      ,							__UpperCamelCase      ,							size=__UpperCamelCase     ) for image in images]

										if do_resize:
														__magic_name__				:					List[Any]						       =       [self.resize(image=__UpperCamelCase      ,							size=__UpperCamelCase      ,							resample=__UpperCamelCase     ) for image in images]

										if do_rescale:
														__magic_name__				:					str						       =       [self.rescale(image=__UpperCamelCase      ,							scale=__UpperCamelCase     ) for image in images]

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

										__magic_name__				:					List[Any]						       =       [to_channel_dimension_format(__UpperCamelCase      ,							__UpperCamelCase     ) for image in images]

										__magic_name__				:					Optional[int]						       =       {"pixel_values": images}
										return BatchFeature(data=__UpperCamelCase      ,							tensor_type=__UpperCamelCase     )