





"""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

A_     :      Tuple								=logging.get_logger(__name__)





def SCREAMING_SNAKE_CASE_   (       snake_case   :					List[str]						,  snake_case   :					Tuple		)->		np.array:
       _lowerCamelCase    	=					f'{sampling_rate}'
       _lowerCamelCase    	=					'1'
       _lowerCamelCase    	=					'f32le'
       _lowerCamelCase    	=					[
           'ffmpeg',
           '-i',
           'pipe:0',
           '-ac',
           ac,
           '-ar',
           ar,
           '-f',
           format_for_conversion,
           '-hide_banner',
           '-loglevel',
           'quiet',
           'pipe:1',
       ]

       try:
              _lowerCamelCase    	=					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'		)
       _lowerCamelCase    	=					ffmpeg_process.communicate(_UpperCAmelCase		)
       _lowerCamelCase    	=					output_stream[0]

       _lowerCamelCase    	=					np.frombuffer(_UpperCAmelCase						,  np.floataa		)
       if audio.shape[0] == 0:
              raise ValueError('Malformed soundfile'		)
       return audio







@add_end_docstrings(lowerCAmelCase__		)
class 	__a   (  lowerCAmelCase__		):



     def __init__(       self					,     *a__					,     **a__   ):
            # Default, might be overriden by the model.config.
            _lowerCamelCase    	=					5
            super().__init__(*UpperCAmelCase_					,     **UpperCAmelCase_   )

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

            self.check_model_type(UpperCAmelCase_   )



     def __call__(       self					,     a__					,     **a__					,     ):
            return super().__call__(UpperCAmelCase_					,     **UpperCAmelCase_   )



     def 						snake_case_			(       self					,     a__=None					,     **a__   ):
            # No parameters on this pipeline right now
            _lowerCamelCase    	=					{}
            if top_k is not None:
                   if top_k > self.model.config.num_labels:
                          _lowerCamelCase    	=					self.model.config.num_labels
                   _lowerCamelCase    	=					top_k
            return {}, {}, postprocess_params



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

            if isinstance(UpperCAmelCase_					,     UpperCAmelCase_   ):
                   _lowerCamelCase    	=					ffmpeg_read(UpperCAmelCase_					,     self.feature_extractor.sampling_rate   )

            if isinstance(UpperCAmelCase_					,     UpperCAmelCase_   ):
                   # 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'   )

                   _lowerCamelCase    	=					inputs.pop('raw'					,     UpperCAmelCase_   )
                   if _inputs is None:
                          # Remove path which will not be used from `datasets`.
                          inputs.pop('path'					,     UpperCAmelCase_   )
                          _lowerCamelCase    	=					inputs.pop('array'					,     UpperCAmelCase_   )
                   _lowerCamelCase    	=					inputs.pop('sampling_rate'   )
                   _lowerCamelCase    	=					_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`.'   )

                          _lowerCamelCase    	=					F.resample(
                              torch.from_numpy(UpperCAmelCase_   )					,     UpperCAmelCase_					,     self.feature_extractor.sampling_rate   ).numpy()

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

            _lowerCamelCase    	=					self.feature_extractor(
                UpperCAmelCase_					,     sampling_rate=self.feature_extractor.sampling_rate					,     return_tensors='pt'   )
            return processed



     def 						snake_case_			(       self					,     a__   ):
            _lowerCamelCase    	=					self.model(**UpperCAmelCase_   )
            return model_outputs



     def 						snake_case_			(       self					,     a__					,     a__=5   ):
            _lowerCamelCase    	=					model_outputs.logits[0].softmax(-1   )
            _lowerCamelCase	,	_lowerCamelCase    	=					probs.topk(UpperCAmelCase_   )

            _lowerCamelCase    	=					scores.tolist()
            _lowerCamelCase    	=					ids.tolist()

            _lowerCamelCase    	=					[{'score': score, 'label': self.model.config.idalabel[_id]} for score, _id in zip(UpperCAmelCase_					,     UpperCAmelCase_   )]

            return labels













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,
    convert_to_rgb,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    OPENAI_CLIP_MEAN,
    OPENAI_CLIP_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


UpperCAmelCase__  :	Dict     =logging.get_logger(__name__)


if is_vision_available():
							import PIL

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


				def __init__(						self      ,       UpperCAmelCase_ = True      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = PILImageResampling.BICUBIC      ,       UpperCAmelCase_ = True      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = True      ,       UpperCAmelCase_ = 1 / 255      ,       UpperCAmelCase_ = True      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = True      ,       **UpperCAmelCase_      ,       ):
								super().__init__(**UpperCAmelCase_							)
								lowerCamelCase		       =size if size is not None else {"""shortest_edge""": 224}
								lowerCamelCase		       =get_size_dict(UpperCAmelCase_      ,       default_to_square=UpperCAmelCase_							)
								lowerCamelCase		       =crop_size if crop_size is not None else {"""height""": 224, """width""": 224}
								lowerCamelCase		       =get_size_dict(UpperCAmelCase_							)

								lowerCamelCase		       =do_resize
								lowerCamelCase		       =size
								lowerCamelCase		       =resample
								lowerCamelCase		       =do_center_crop
								lowerCamelCase		       =crop_size
								lowerCamelCase		       =do_rescale
								lowerCamelCase		       =rescale_factor
								lowerCamelCase		       =do_normalize
								lowerCamelCase		       =image_mean if image_mean is not None else OPENAI_CLIP_MEAN
								lowerCamelCase		       =image_std if image_std is not None else OPENAI_CLIP_STD
								lowerCamelCase		       =do_convert_rgb


				def _snake_case				(						self      ,       UpperCAmelCase_      ,       UpperCAmelCase_      ,       UpperCAmelCase_ = PILImageResampling.BICUBIC      ,       UpperCAmelCase_ = None      ,       **UpperCAmelCase_      ,       ):
								lowerCamelCase		       =get_size_dict(UpperCAmelCase_      ,       default_to_square=UpperCAmelCase_							)
								lowerCamelCase		       =get_resize_output_image_size(
								    UpperCAmelCase_      ,       size=(size["""height"""], size["""width"""])      ,       default_to_square=UpperCAmelCase_							)
								return resize(UpperCAmelCase_      ,       size=UpperCAmelCase_      ,       resample=UpperCAmelCase_      ,       data_format=UpperCAmelCase_      ,       **UpperCAmelCase_							)


				def _snake_case				(						self      ,       UpperCAmelCase_      ,       UpperCAmelCase_      ,       UpperCAmelCase_ = None      ,       **UpperCAmelCase_      ,       ):
								lowerCamelCase		       =get_size_dict(UpperCAmelCase_							)
								return center_crop(UpperCAmelCase_      ,       size=(size["""height"""], size["""width"""])      ,       data_format=UpperCAmelCase_      ,       **UpperCAmelCase_							)


				def _snake_case				(						self      ,       UpperCAmelCase_      ,       UpperCAmelCase_      ,       UpperCAmelCase_ = None      ,       **UpperCAmelCase_      ,       ):
								return rescale(UpperCAmelCase_      ,       scale=UpperCAmelCase_      ,       data_format=UpperCAmelCase_      ,       **UpperCAmelCase_							)


				def _snake_case				(						self      ,       UpperCAmelCase_      ,       UpperCAmelCase_      ,       UpperCAmelCase_      ,       UpperCAmelCase_ = None      ,       **UpperCAmelCase_      ,       ):
								return normalize(UpperCAmelCase_      ,       mean=UpperCAmelCase_      ,       std=UpperCAmelCase_      ,       data_format=UpperCAmelCase_      ,       **UpperCAmelCase_							)


				def _snake_case				(						self      ,       UpperCAmelCase_      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = None      ,       UpperCAmelCase_ = ChannelDimension.FIRST      ,       **UpperCAmelCase_      ,       ):
								lowerCamelCase		       =do_resize if do_resize is not None else self.do_resize
								lowerCamelCase		       =size if size is not None else self.size
								lowerCamelCase		       =get_size_dict(UpperCAmelCase_      ,       default_to_square=UpperCAmelCase_							)
								lowerCamelCase		       =resample if resample is not None else self.resample
								lowerCamelCase		       =do_center_crop if do_center_crop is not None else self.do_center_crop
								lowerCamelCase		       =crop_size if crop_size is not None else self.crop_size
								lowerCamelCase		       =get_size_dict(UpperCAmelCase_							)
								lowerCamelCase		       =do_rescale if do_rescale is not None else self.do_rescale
								lowerCamelCase		       =rescale_factor if rescale_factor is not None else self.rescale_factor
								lowerCamelCase		       =do_normalize if do_normalize is not None else self.do_normalize
								lowerCamelCase		       =image_mean if image_mean is not None else self.image_mean
								lowerCamelCase		       =image_std if image_std is not None else self.image_std
								lowerCamelCase		       =do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

								lowerCamelCase		       =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:
												raise ValueError("""Size 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."""							)

								# PIL RGBA images are converted to RGB
								if do_convert_rgb:
												lowerCamelCase		       =[convert_to_rgb(UpperCAmelCase_							) for image in images]

								# All transformations expect numpy arrays.
								lowerCamelCase		       =[to_numpy_array(UpperCAmelCase_							) for image in images]

								if do_resize:
												lowerCamelCase		       =[self.resize(image=UpperCAmelCase_      ,       size=UpperCAmelCase_      ,       resample=UpperCAmelCase_							) for image in images]

								if do_center_crop:
												lowerCamelCase		       =[self.center_crop(image=UpperCAmelCase_      ,       size=UpperCAmelCase_							) for image in images]

								if do_rescale:
												lowerCamelCase		       =[self.rescale(image=UpperCAmelCase_      ,       scale=UpperCAmelCase_							) for image in images]

								if do_normalize:
												lowerCamelCase		       =[self.normalize(image=UpperCAmelCase_      ,       mean=UpperCAmelCase_      ,       std=UpperCAmelCase_							) for image in images]

								lowerCamelCase		       =[to_channel_dimension_format(UpperCAmelCase_      ,       UpperCAmelCase_							) for image in images]

								lowerCamelCase		       ={"""pixel_values""": images}
								return BatchFeature(data=UpperCAmelCase_      ,       tensor_type=UpperCAmelCase_							)





