





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

__lowercase				    =      logging.get_logger(__name__)


def        __SCREAMING_SNAKE_CASE	(				__lowercase:		bytes  ,				__lowercase:		int):



  '''simple docstring'''






  __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(_a )
class 	__snake_case					(      _a ):






    '''simple docstring'''





    def __init__( self    :		Any      ,    *snake_case    :		Union[str, Any]      ,    **snake_case    :		Dict				):

      """simple docstring"""





      __lowerCAmelCase								=					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    :		Any      ,    snake_case    :		Union[np.ndarray, bytes, str]      ,    **snake_case    :		List[Any]      ,    ):

      """simple docstring"""





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





    def 						A		( self    :		int      ,    snake_case    :		List[str]=None      ,    **snake_case    :		Dict				):

      """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      ,    snake_case    :		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
          __lowerCAmelCase								=					requests.get(snake_case				).content
        else:
          with open(snake_case      ,    '''rb'''				) as f:
            __lowerCAmelCase								=					f.read()

      if isinstance(snake_case      ,    snake_case				):
        __lowerCAmelCase								=					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'''				)

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

      __lowerCAmelCase								=					self.feature_extractor(
          snake_case      ,    sampling_rate=self.feature_extractor.sampling_rate      ,    return_tensors='''pt'''				)
      return processed





    def 						A		( self    :		Optional[int]      ,    snake_case    :		Tuple				):

      """simple docstring"""





      __lowerCAmelCase								=					self.model(**snake_case				)
      return model_outputs





    def 						A		( self    :		int      ,    snake_case    :		Dict      ,    snake_case    :		List[Any]=5				):

      """simple docstring"""





      __lowerCAmelCase								=					model_outputs.logits[0].softmax(-1				)
      __lowerCAmelCase						,    __lowerCAmelCase								=					probs.topk(snake_case				)

      __lowerCAmelCase								=					scores.tolist()
      __lowerCAmelCase								=					ids.tolist()

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

      return labels







import os
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device


if is_torch_available():
      import torch

      from transformers import GPTaLMHeadModel

@require_torch
@require_sentencepiece
@require_tokenizers
class 	__snake_case					(      unittest.TestCase ):






    '''simple docstring'''





    @slow
    @unittest.skip('''Model is not available.'''				)
    def 						A		( self    :		str				):

      """simple docstring"""





      __lowerCAmelCase								=					'''nvidia/megatron-gpt2-345m/'''
      if "MYDIR" in os.environ:
        __lowerCAmelCase								=					os.path.join(os.environ['''MYDIR''']      ,    snake_case				)
      __lowerCAmelCase								=					GPTaLMHeadModel.from_pretrained(snake_case				)
      model.to(snake_case				)
      model.half()

      __lowerCAmelCase								=					torch.tensor(
          [[101, 7_110, 1_005, 1_056, 2_023, 11_333, 17_413, 1_029, 102]]      ,    device=snake_case      ,    dtype=torch.long      ,    )

      with torch.no_grad():
        __lowerCAmelCase								=					model(snake_case				).logits

      __lowerCAmelCase								=					torch.Size((1, 9, 50_257)				)
      self.assertEqual(output.shape      ,    snake_case				)

      __lowerCAmelCase								=					torch.tensor(
          [
              4.9414,
              -0.2920,
              -1.2148,
              -4.0273,
              -0.5161,
              -5.2109,
              -1.2412,
              -1.8301,
              -1.7734,
              -4.7148,
              -0.2317,
              -1.0811,
              -2.1777,
              0.4141,
              -3.7969,
              -4.0586,
              -2.5332,
              -3.3809,
              4.3867,
          ]      ,    device=snake_case      ,    dtype=torch.half      ,    )

      for i in range(19				):
        __lowerCAmelCase						,    __lowerCAmelCase								=					8 * i // 17, 2_792 * i  # along the diagonal
        __lowerCAmelCase						,    __lowerCAmelCase								=					output[0, r, c], expected_diag[i]
        __lowerCAmelCase								=					f'''row={r} col={c} computed={computed} expected={expected}'''
        self.assertAlmostEqual(snake_case      ,    snake_case      ,    delta=1e-4      ,    msg=snake_case				)
