code: 
  
  
  
  
  
  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       = logging.get_logger(__name__)
  
  
  
  
  
  
  
  def       lowerCAmelCase__(	lowercase       :    str   ,     lowercase       :    Optional[int]				) ->    np.array:
    __snake_case   :				Tuple        =       f"""{sampling_rate}"""
    __snake_case   :				int        =       '''1'''
    __snake_case   :				Optional[Any]        =       '''f32le'''
    __snake_case   :				Dict        =       [
        '''ffmpeg''',
        '''-i''',
        '''pipe:0''',
        '''-ac''',
        ac,
        '''-ar''',
        ar,
        '''-f''',
        format_for_conversion,
        '''-hide_banner''',
        '''-loglevel''',
        '''quiet''',
        '''pipe:1''',
    ]
  
    try:
      __snake_case   :				Tuple        =       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"				)
    __snake_case   :				Optional[Any]        =       ffmpeg_process.communicate(lowercase				)
    __snake_case   :				Union[str, Any]        =       output_stream[0]
  
    __snake_case   :				Optional[int]        =       np.frombuffer(lowercase   ,     np.floataa				)
    if audio.shape[0] == 0:
      raise ValueError("Malformed soundfile"				)
    return audio
  
  
  
  
  @add_end_docstrings(a_    )
  class 				_lowerCamelCase (    a_    ):
  
  
  
        """simple docstring"""
  
  
  
  
  
  
        def __init__( self				,						*UpperCAmelCase				,						**UpperCAmelCase   )       ->   Any:
  
  
  
  
  
  
          '''simple docstring'''
  
  
  
  
  
  
          __snake_case   :				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				,						UpperCAmelCase				,						**UpperCAmelCase				,						)       ->   int:
  
  
  
  
  
  
          '''simple docstring'''
  
  
  
  
  
  
          return super().__call__(_snake_case				,						**_snake_case   )
  
  
  
  
  
  
  
        def 						UpperCAmelCase		( self				,						UpperCAmelCase=None				,						**UpperCAmelCase   )       ->   Union[str, Any]:
  
  
  
  
  
  
          '''simple docstring'''
  
  
  
  
  
  
          __snake_case   :				List[str]        =       {}
          if top_k is not None:
            if top_k > self.model.config.num_labels:
              __snake_case   :				int        =       self.model.config.num_labels
            __snake_case   :				Tuple        =       top_k
          return {}, {}, postprocess_params
  
  
  
  
  
  
  
        def 						UpperCAmelCase		( self				,						UpperCAmelCase   )       ->   Optional[Any]:
  
  
  
  
  
  
          '''simple docstring'''
  
  
  
  
  
  
          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
              __snake_case   :				Dict        =       requests.get(_snake_case   ).content
            else:
              with open(_snake_case				,						"rb"   ) as f:
                __snake_case   :				int        =       f.read()
  
          if isinstance(_snake_case				,						_snake_case   ):
            __snake_case   :				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"   )
  
            __snake_case   :				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   )
              __snake_case   :				str        =       inputs.pop("array"				,						_snake_case   )
            __snake_case   :				Optional[Any]        =       inputs.pop("sampling_rate"   )
            __snake_case   :				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`."   )
  
              __snake_case   :				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"   )
  
          __snake_case   :				List[Any]        =       self.feature_extractor(
              _snake_case				,						sampling_rate=self.feature_extractor.sampling_rate				,						return_tensors="pt"   )
          return processed
  
  
  
  
  
  
  
        def 						UpperCAmelCase		( self				,						UpperCAmelCase   )       ->   str:
  
  
  
  
  
  
          '''simple docstring'''
  
  
  
  
  
  
          __snake_case   :				Tuple        =       self.model(**_snake_case   )
          return model_outputs
  
  
  
  
  
  
  
        def 						UpperCAmelCase		( self				,						UpperCAmelCase				,						UpperCAmelCase=5   )       ->   int:
  
  
  
  
  
  
          '''simple docstring'''
  
  
  
  
  
  
          __snake_case   :				str        =       model_outputs.logits[0].softmax(-1   )
          __snake_case   :				Union[str, Any]        =       probs.topk(_snake_case   )
  
          __snake_case   :				Dict        =       scores.tolist()
          __snake_case   :				Dict        =       ids.tolist()
  
          __snake_case   :				Any        =       [{'''score''': score, '''label''': self.model.config.idalabel[_id]} for score, _id in zip(_snake_case				,						_snake_case   )]
  
          return labels
  
  
style_context: 
  
  
  
  
  
  from typing import TYPE_CHECKING
  
  from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available
  
  
  _UpperCamelCase       = {'''configuration_nat''': ['''NAT_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''NatConfig''']}
  
  
  try:
       if not is_torch_available():
            raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
       pass
  else:
       _UpperCamelCase       = [
           '''NAT_PRETRAINED_MODEL_ARCHIVE_LIST''',
           '''NatForImageClassification''',
           '''NatModel''',
           '''NatPreTrainedModel''',
           '''NatBackbone''',
       ]
  
  if TYPE_CHECKING:
       from .configuration_nat import NAT_PRETRAINED_CONFIG_ARCHIVE_MAP, NatConfig
  
       try:
            if not is_torch_available():
                 raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
            pass
       else:
            from .modeling_nat import (
                NAT_PRETRAINED_MODEL_ARCHIVE_LIST,
                NatBackbone,
                NatForImageClassification,
                NatModel,
                NatPreTrainedModel,
            )
  
  else:
       import sys
  
       _UpperCamelCase       = _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)
  
  