






'''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				:				int					=		logging.get_logger(__name__)






def  lowerCamelCase_  (     lowercase__					,  lowercase__):
 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(lowercase__					,  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(lowercase__)
 lowerCamelCase__	=  output_stream[0]

 lowerCamelCase__	=  np.frombuffer(lowercase__					,  np.floataa)
 if audio.shape[0] == 0:
  raise ValueError("Malformed soundfile")
 return audio
@add_end_docstrings(_lowerCamelCase   )
class 					lowercase							(				_lowerCamelCase   ):


    '''simple docstring'''

    def __init__(       self  :  Optional[int]					,  *__lowerCamelCase  :  Union[str, Any]					,  **__lowerCamelCase  :  str  )				->				int:

     '''simple docstring'''




     lowerCamelCase__	=  5
     super().__init__(*__lowerCamelCase					,  **__lowerCamelCase  )

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

     self.check_model_type(__lowerCamelCase  )

    def __call__(       self  :  List[Any]					,  __lowerCamelCase  :  Union[np.ndarray, bytes, str]					,  **__lowerCamelCase  :  Union[str, Any]					,  )				->				str:

     '''simple docstring'''




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

    def  a__					(       self  :  Tuple					,  __lowerCamelCase  :  Dict=None					,  **__lowerCamelCase  :  str  )				->				List[Any]:

     '''simple docstring'''




     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  a__					(       self  :  Any					,  __lowerCamelCase  :  int  )				->				Any:

     '''simple docstring'''




     if isinstance(__lowerCamelCase					,  __lowerCamelCase  ):
      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(__lowerCamelCase  ).content
      else:
       with open(__lowerCamelCase					,  "rb"  ) as f:
        lowerCamelCase__	=  f.read()

     if isinstance(__lowerCamelCase					,  __lowerCamelCase  ):
      lowerCamelCase__	=  ffmpeg_read(__lowerCamelCase					,  self.feature_extractor.sampling_rate  )

     if isinstance(__lowerCamelCase					,  __lowerCamelCase  ):
      # 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"					,  __lowerCamelCase  )
      if _inputs is None:
       # Remove path which will not be used from `datasets`.
       inputs.pop("path"					,  __lowerCamelCase  )
       lowerCamelCase__	=  inputs.pop("array"					,  __lowerCamelCase  )
      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(__lowerCamelCase  )					,  __lowerCamelCase					,  self.feature_extractor.sampling_rate  ).numpy()

     if not isinstance(__lowerCamelCase					,  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(
         __lowerCamelCase					,  sampling_rate=self.feature_extractor.sampling_rate					,  return_tensors="pt"  )
     return processed

    def  a__					(       self  :  str					,  __lowerCamelCase  :  Tuple  )				->				str:

     '''simple docstring'''




     lowerCamelCase__	=  self.model(**__lowerCamelCase  )
     return model_outputs




    def  a__					(       self  :  Tuple					,  __lowerCamelCase  :  List[Any]					,  __lowerCamelCase  :  List[str]=5  )				->				Optional[int]:

     '''simple docstring'''




     lowerCamelCase__	=  model_outputs.logits[0].softmax(-1  )
     lowerCamelCase__	,						lowerCamelCase__	=  probs.topk(__lowerCamelCase  )

     lowerCamelCase__	=  scores.tolist()
     lowerCamelCase__	=  ids.tolist()

     lowerCamelCase__	=  [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(__lowerCamelCase					,  __lowerCamelCase  )]

     return labels









'''simple docstring'''





def  lowerCamelCase_  (     lowercase__):
 return int(lowercase__) if x - int(lowercase__) <= 0 else int(lowercase__) + 1


if __name__ == "__main__":
 import doctest

 doctest.testmod()

