


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


def 					__lowerCAmelCase    (    a__		,					a__						) ->						np.array:
				__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(a__		,					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(a__						)
				__a    							=					output_stream[0]

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







@add_end_docstrings(a    )
class     __A(				a    ):




				def __init__(  self	,  *_snake_case	,  **_snake_case       )					->   List[Any]:


								'''simple docstring'''




								__a    							=					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	,  )					->   Optional[Any]:


								'''simple docstring'''




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




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case=None	,  **_snake_case       )					->   int:


								'''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        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Union[str, 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
																__a    							=					requests.get(_snake_case       ).content
												else:
																with open(_snake_case	,  '''rb'''       ) as f:
																				__a    							=					f.read()

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

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

								__a    							=					self.feature_extractor(
								    _snake_case	,  sampling_rate=self.feature_extractor.sampling_rate	,  return_tensors='''pt'''       )
								return processed




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   List[Any]:


								'''simple docstring'''




								__a    							=					self.model(**_snake_case       )
								return model_outputs




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case=5       )					->   int:


								'''simple docstring'''




								__a    							=					model_outputs.logits[0].softmax(-1       )
								__a							,  __a    							=					probs.topk(_snake_case       )

								__a    							=					scores.tolist()
								__a    							=					ids.tolist()

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

								return labels

import math
from collections import OrderedDict

import torch
from packaging import version
from torch import Tensor, nn

from .utils import logging


A  :			List[str]     				=				logging.get_logger(__name__)







class     __A(				nn.Module    ):




				def __init__(  self       )					->   Union[str, Any]:


								'''simple docstring'''




								super().__init__()
								if version.parse(torch.__version__       ) < version.parse('''1.12.0'''       ):
												raise ImportError(
												    F"""You are using torch=={torch.__version__}, but torch>=1.12.0 is required to use """
												    '''PytorchGELUTanh. Please upgrade torch.'''       )




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return nn.functional.gelu(_snake_case	,  approximate='''tanh'''       )







class     __A(				nn.Module    ):




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return 0.5 * input * (1.0 + torch.tanh(math.sqrt(2.0 / math.pi       ) * (input + 0.04_4715 * torch.pow(_snake_case	,  3.0       ))       ))







class     __A(				nn.Module    ):




				def __init__(  self	,  _snake_case = False       )					->   Tuple:


								'''simple docstring'''




								super().__init__()
								if use_gelu_python:
												__a    							=					self._gelu_python
								else:
												__a    							=					nn.functional.gelu




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return input * 0.5 * (1.0 + torch.erf(input / math.sqrt(2.0       )       ))




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return self.act(_snake_case       )







class     __A(				nn.Module    ):




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return 0.5 * input * (1.0 + torch.tanh(input * 0.79_7884_5608 * (1.0 + 0.04_4715 * input * input)       ))







class     __A(				nn.Module    ):




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return input * torch.sigmoid(1.702 * input       )







class     __A(				nn.Module    ):




				def __init__(  self	,  _snake_case	,  _snake_case       )					->   Optional[int]:


								'''simple docstring'''




								if min > max:
												raise ValueError(F"""min should be < max (got min: {min}, max: {max})"""       )

								super().__init__()
								__a    							=					min
								__a    							=					max




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return torch.clip(gelu(_snake_case       )	,  self.min	,  self.max       )







class     __A(				nn.Module    ):




				def __init__(  self       )					->   str:


								'''simple docstring'''




								super().__init__()
								__a    							=					math.sqrt(2 / math.pi       )




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return 0.5 * input * (1 + torch.tanh(self.precomputed_constant * (input + 0.04_4715 * torch.pow(_snake_case	,  3       ))       ))







class     __A(				nn.Module    ):




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return nn.functional.silu(_snake_case       )







class     __A(				nn.Module    ):




				def __init__(  self       )					->   List[Any]:


								'''simple docstring'''




								super().__init__()
								if version.parse(torch.__version__       ) < version.parse('''1.9.0'''       ):
												__a    							=					self._mish_python
								else:
												__a    							=					nn.functional.mish




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return input * torch.tanh(nn.functional.softplus(_snake_case       )       )




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return self.act(_snake_case       )







class     __A(				nn.Module    ):




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Tensor:


								'''simple docstring'''




								return input







class     __A(				nn.Module    ):




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case=0.70_7107	,  _snake_case=0.28_2095       )					->   int:


								'''simple docstring'''




								__a    							=					(input - mu).div(sigma * math.sqrt(2.0       )       )
								return 0.5 * (1.0 + torch.erf(_snake_case       ))







class     __A(				nn.Module    ):




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Any:


								'''simple docstring'''




								__a    							=					nn.functional.relu(_snake_case       )
								__a    							=					torch.square(_snake_case       )
								return squared







class     __A(				a    ):




				def __getitem__(  self	,  _snake_case       )					->   Dict:


								'''simple docstring'''




								__a    							=					super().__getitem__(_snake_case       )
								__a							,  __a    							=					content if isinstance(_snake_case	,  _snake_case       ) else (content, {})
								return cls(**_snake_case       )



A  :			str     				=				{
    'gelu': GELUActivation,
    'gelu_10': (ClippedGELUActivation, {'min': -1_0, 'max': 1_0}),
    'gelu_fast': FastGELUActivation,
    'gelu_new': NewGELUActivation,
    'gelu_python': (GELUActivation, {'use_gelu_python': True}),
    'gelu_pytorch_tanh': PytorchGELUTanh,
    'gelu_accurate': AccurateGELUActivation,
    'laplace': LaplaceActivation,
    'linear': LinearActivation,
    'mish': MishActivation,
    'quick_gelu': QuickGELUActivation,
    'relu': nn.ReLU,
    'relu2': ReLUSquaredActivation,
    'relu6': nn.ReLUa,
    'sigmoid': nn.Sigmoid,
    'silu': SiLUActivation,
    'swish': SiLUActivation,
    'tanh': nn.Tanh,
}
A  :			Any     				=				ClassInstantier(ACTaCLS)


def 					__lowerCAmelCase    (    a__						) ->						Optional[int]:
				if activation_string in ACTaFN:
								return ACTaFN[activation_string]
				else:
								raise KeyError(F"""function {activation_string} not found in ACT2FN mapping {list(ACTaFN.keys()						)}"""						)


# For backwards compatibility with: from activations import gelu_python
A  :			Optional[Any]     				=				get_activation('gelu_python')
A  :			str     				=				get_activation('gelu_new')
A  :			Any     				=				get_activation('gelu')
A  :			int     				=				get_activation('gelu_fast')
A  :			Any     				=				get_activation('quick_gelu')
A  :			int     				=				get_activation('silu')
A  :			List[str]     				=				get_activation('mish')






A  :			Dict     				=				get_activation('linear')