code: 
  '''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
  
  _UpperCAmelCase   :      int 						=  logging.get_logger(__name__)
  
  
  
  
  
  
  
  def 						snake_case__	(      UpperCamelCase			,UpperCamelCase	) ->      np.array:
  		_UpperCamelCase       : Tuple    =      f'''{sampling_rate}'''
  		_UpperCamelCase       : int    =      '''1'''
  		_UpperCamelCase       : Optional[Any]    =      '''f32le'''
  		_UpperCamelCase       : Dict    =      [
  		    '''ffmpeg''',
  		    '''-i''',
  		    '''pipe:0''',
  		    '''-ac''',
  		    ac,
  		    '''-ar''',
  		    ar,
  		    '''-f''',
  		    format_for_conversion,
  		    '''-hide_banner''',
  		    '''-loglevel''',
  		    '''quiet''',
  		    '''pipe:1''',
  		]
  
  		try:
  				_UpperCamelCase       : Tuple    =      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'''	)
  		_UpperCamelCase       : Optional[Any]    =      ffmpeg_process.communicate(UpperCamelCase	)
  		_UpperCamelCase       : Union[str, Any]    =      output_stream[0]
  
  		_UpperCamelCase       : Optional[int]    =      np.frombuffer(UpperCamelCase			,np.floataa	)
  		if audio.shape[0] == 0:
  				raise ValueError('''Malformed soundfile'''	)
  		return audio
  
  
  @add_end_docstrings(a_ )
  class 			UpperCAmelCase				(					a_ ):
  
  
  
  
  
  		"""simple docstring"""
  
  
  
  
  
  		def __init__(    self  ,					*_snake_case  ,					**_snake_case     )				->    Any:
  				# Default, might be overriden by the model.config.
  				_UpperCamelCase       : List[str]    =      5
  				super().__init__(*_snake_case  ,					**_snake_case     )
  
  				if self.framework != "pt":
  						raise ValueError(F'''The {self.__class__} is only available in PyTorch.'''     )
  
  				self.check_model_type(_snake_case     )
  		def __call__(    self  ,					_snake_case  ,					**_snake_case  ,					)				->    int:
  				return super().__call__(_snake_case  ,					**_snake_case     )
  		def _lowercase  (    self  ,					_snake_case=None  ,					**_snake_case     )				->    Union[str, Any]:
  				# No parameters on this pipeline right now
  				_UpperCamelCase       : List[str]    =      {}
  				if top_k is not None:
  						if top_k > self.model.config.num_labels:
  								_UpperCamelCase       : int    =      self.model.config.num_labels
  						_UpperCamelCase       : Tuple    =      top_k
  				return {}, {}, postprocess_params
  		def _lowercase  (    self  ,					_snake_case     )				->    Optional[Any]:
  				if isinstance(_snake_case  ,					_snake_case     ):
  						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
  								_UpperCamelCase       : Dict    =      requests.get(_snake_case     ).content
  						else:
  								with open(_snake_case  ,					'''rb'''     ) as f:
  										_UpperCamelCase       : int    =      f.read()
  
  				if isinstance(_snake_case  ,					_snake_case     ):
  						_UpperCamelCase       : str    =      ffmpeg_read(_snake_case  ,					self.feature_extractor.sampling_rate     )
  
  				if isinstance(_snake_case  ,					_snake_case     ):
  						# 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'''     )
  
  						_UpperCamelCase       : List[Any]    =      inputs.pop('''raw'''  ,					_snake_case     )
  						if _inputs is None:
  								# Remove path which will not be used from `datasets`.
  								inputs.pop('''path'''  ,					_snake_case     )
  								_UpperCamelCase       : str    =      inputs.pop('''array'''  ,					_snake_case     )
  						_UpperCamelCase       : Optional[Any]    =      inputs.pop('''sampling_rate'''     )
  						_UpperCamelCase       : 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`.'''     )
  
  								_UpperCamelCase       : Tuple    =      F.resample(
  								    torch.from_numpy(_snake_case     )  ,					_snake_case  ,					self.feature_extractor.sampling_rate     ).numpy()
  
  				if not isinstance(_snake_case  ,					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'''     )
  
  				_UpperCamelCase       : List[Any]    =      self.feature_extractor(
  				    _snake_case  ,					sampling_rate=self.feature_extractor.sampling_rate  ,					return_tensors='''pt'''     )
  				return processed
  		def _lowercase  (    self  ,					_snake_case     )				->    str:
  				_UpperCamelCase       : Tuple    =      self.model(**_snake_case     )
  				return model_outputs
  
  
  
  
  		def _lowercase  (    self  ,					_snake_case  ,					_snake_case=5     )				->    int:
  				_UpperCamelCase       : str    =      model_outputs.logits[0].softmax(-1     )
  				_UpperCamelCase,							_UpperCamelCase       : Union[str, Any]    =      probs.topk(_snake_case     )
  
  				_UpperCamelCase       : Dict    =      scores.tolist()
  				_UpperCamelCase       : Dict    =      ids.tolist()
  
  				_UpperCamelCase       : Any    =      [{'''score''': score, '''label''': self.model.config.idalabel[_id]} for score, _id in zip(_snake_case  ,					_snake_case     )]
  
  				return labels
  
  
  
  
  
  
style_context: 
  '''simple docstring'''
  
  _UpperCAmelCase   :      Optional[int] 						=  """
  # Transformers installation
  ! pip install transformers datasets evaluate
  # To install from source instead of the last release, comment the command above and uncomment the following one.
  # ! pip install git+https://github.com/huggingface/transformers.git
  """
  
  _UpperCAmelCase   :      Union[str, Any] 						=  [{"""type""": """code""", """content""": INSTALL_CONTENT}]
  _UpperCAmelCase   :      Union[str, Any] 						=  {
      """{processor_class}""": """FakeProcessorClass""",
      """{model_class}""": """FakeModelClass""",
      """{object_class}""": """FakeObjectClass""",
  }
  
  
  
  
  
  