






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 		_a					(   a				:Optional[int]			,							a				:List[str]				)							->    Optional[Any]:
 a							=   F"""{sampling_rate}"""
 a							=   '''1'''
 a							=   '''f32le'''
 a							=   [
     '''ffmpeg''',
     '''-i''',
     '''pipe:0''',
     '''-ac''',
     ac,
     '''-ar''',
     ar,
     '''-f''',
     format_for_conversion,
     '''-hide_banner''',
     '''-loglevel''',
     '''quiet''',
     '''pipe:1''',
 ]

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

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






@add_end_docstrings(lowercase     )
class 	lowercase_      (			lowercase     ):






      '''simple docstring'''




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




       """simple docstring"""


       a							=   5
       super().__init__(*a_				,      **a_						)

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

       self.check_model_type(a_						)







      def __call__(   self   :		Any				,      __UpperCAmelCase   :		List[Any]				,      **__UpperCAmelCase   :		Tuple				,      )					->int:




       """simple docstring"""


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







      def    __lowerCAmelCase      (   self   :		Any				,      __UpperCAmelCase   :		str=None				,      **__UpperCAmelCase   :		Union[str, Any]						)					->Dict:




       """simple docstring"""


       a							=   {}
       if top_k is not None:
        if top_k > self.model.config.num_labels:
         a							=   self.model.config.num_labels
        a							=   top_k
       return {}, {}, postprocess_params







      def    __lowerCAmelCase      (   self   :		List[Any]				,      __UpperCAmelCase   :		Optional[int]						)					->str:




       """simple docstring"""


       if isinstance(a_				,      a_						):
        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
         a							=   requests.get(a_						).content
        else:
         with open(a_				,      '''rb'''						) as f:
          a							=   f.read()

       if isinstance(a_				,      a_						):
        a							=   ffmpeg_read(a_				,      self.feature_extractor.sampling_rate						)

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

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

         a							=   F.resample(
             torch.from_numpy(a_						)				,      a_				,      self.feature_extractor.sampling_rate						).numpy()

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

       a							=   self.feature_extractor(
           a_				,      sampling_rate=self.feature_extractor.sampling_rate				,      return_tensors='''pt'''						)
       return processed







      def    __lowerCAmelCase      (   self   :		int				,      __UpperCAmelCase   :		Tuple						)					->List[str]:




       """simple docstring"""


       a							=   self.model(**a_						)
       return model_outputs







      def    __lowerCAmelCase      (   self   :		int				,      __UpperCAmelCase   :		Optional[int]				,      __UpperCAmelCase   :		Union[str, Any]=5						)					->int:




       """simple docstring"""


       a							=   model_outputs.logits[0].softmax(-1						)
       a						,  a							=   probs.topk(a_						)

       a							=   scores.tolist()
       a							=   ids.tolist()

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

       return labels










import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding






class 	lowercase_      (			lowercase     ):






      '''simple docstring'''




      __snake_case  						=						['''image_processor''', '''tokenizer''']
      __snake_case  						=						'''AutoImageProcessor'''
      __snake_case  						=						'''AutoTokenizer'''







      def __init__(   self   :		Any				,      __UpperCAmelCase   :		Union[str, Any]=None				,      __UpperCAmelCase   :		int=None				,      **__UpperCAmelCase   :		Union[str, Any]						)					->Dict:




       """simple docstring"""


       a							=   None
       if "feature_extractor" in kwargs:
        warnings.warn(
            '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
            ''' instead.'''				,      __UpperCAmelCase				,      )
        a							=   kwargs.pop('''feature_extractor'''						)

       a							=   image_processor if image_processor is not None else feature_extractor
       if image_processor is None:
        raise ValueError('''You have to specify an image_processor.'''						)
       if tokenizer is None:
        raise ValueError('''You have to specify a tokenizer.'''						)

       super().__init__(__UpperCAmelCase				,      __UpperCAmelCase						)
       a							=   self.image_processor







      def __call__(   self   :		List[Any]				,      __UpperCAmelCase   :		Optional[Any]=None				,      __UpperCAmelCase   :		Optional[int]=None				,      __UpperCAmelCase   :		str=None				,      **__UpperCAmelCase   :		List[Any]						)					->Dict:




       """simple docstring"""


       if text is None and images is None:
        raise ValueError('''You have to specify either text or images. Both cannot be none.'''						)

       if text is not None:
        a							=   self.tokenizer(__UpperCAmelCase				,      return_tensors=__UpperCAmelCase				,      **__UpperCAmelCase						)

       if images is not None:
        a							=   self.image_processor(__UpperCAmelCase				,      return_tensors=__UpperCAmelCase				,      **__UpperCAmelCase						)

       if text is not None and images is not None:
        a							=   image_features.pixel_values
        return encoding
       elif text is not None:
        return encoding
       else:
        return BatchEncoding(data=dict(**__UpperCAmelCase						)				,      tensor_type=__UpperCAmelCase						)







      def    __lowerCAmelCase      (   self   :		List[Any]				,      *__UpperCAmelCase   :		Optional[Any]				,      **__UpperCAmelCase   :		List[str]						)					->Optional[Any]:




       """simple docstring"""


       return self.tokenizer.batch_decode(*__UpperCAmelCase				,      **__UpperCAmelCase						)







      def    __lowerCAmelCase      (   self   :		str				,      *__UpperCAmelCase   :		List[Any]				,      **__UpperCAmelCase   :		Optional[Any]						)					->Union[str, Any]:




       """simple docstring"""


       return self.tokenizer.decode(*__UpperCAmelCase				,      **__UpperCAmelCase						)







      @property
      def    __lowerCAmelCase      (   self   :		str						)					->str:




       """simple docstring"""


       a							=   self.tokenizer.model_input_names
       a							=   self.image_processor.model_input_names
       return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names						)						)







      @property
      def    __lowerCAmelCase      (   self   :		str						)					->Optional[Any]:




       """simple docstring"""


       warnings.warn(
           '''`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'''				,      __UpperCAmelCase				,      )
       return self.image_processor_class







      @property
      def    __lowerCAmelCase      (   self   :		int						)					->Optional[Any]:




       """simple docstring"""


       warnings.warn(
           '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''				,      __UpperCAmelCase				,      )
       return self.image_processor


