
"""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

lowerCAmelCase								=						logging.get_logger(__name__)






def        lowerCAmelCase_       (		snake_case_		:      bytes							,   snake_case_		:      int      )       ->np.array:
			lowerCamelCase__     :					Union[str, Any]		=f"""{sampling_rate}"""
			lowerCamelCase__     :					Tuple		='1'
			lowerCamelCase__     :					Union[str, Any]		='f32le'
			lowerCamelCase__     :					Dict		=[
			    'ffmpeg',
			    '-i',
			    'pipe:0',
			    '-ac',
			    ac,
			    '-ar',
			    ar,
			    '-f',
			    format_for_conversion,
			    '-hide_banner',
			    '-loglevel',
			    'quiet',
			    'pipe:1',
			]

			try:
						lowerCamelCase__     :					Dict		=subprocess.Popen(snake_case_							,   stdin=subprocess.PIPE							,   stdout=subprocess.PIPE      )
			except FileNotFoundError:
						raise ValueError('ffmpeg was not found but is required to load audio files from filename'      )
			lowerCamelCase__     :					Optional[Any]		=ffmpeg_process.communicate(snake_case_      )
			lowerCamelCase__     :					Optional[int]		=output_stream[0]

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

@add_end_docstrings(A__       )
class    A_     (					A__       ):



							"""simple docstring"""






							def __init__(  self  :Tuple					,							*lowerCamelCase_  :Any					,							**lowerCamelCase_  :str	):





										"""simple docstring"""

										lowerCamelCase__     :					Union[str, Any]		=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_  :Tuple					,							):





										"""simple docstring"""

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





							def 				UpperCAmelCase__							(  self  :Dict					,							lowerCamelCase_  :Union[str, Any]=None					,							**lowerCamelCase_  :Any	):





										"""simple docstring"""

										lowerCamelCase__     :					List[str]		={}
										if top_k is not None:
													if top_k > self.model.config.num_labels:
																lowerCamelCase__     :					List[Any]		=self.model.config.num_labels
													lowerCamelCase__     :					int		=top_k
										return {}, {}, postprocess_params





							def 				UpperCAmelCase__							(  self  :str					,							lowerCamelCase_  :List[str]	):





										"""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__     :					Any		=requests.get(lowerCamelCase_	).content
													else:
																with open(lowerCamelCase_					,							'rb'	) as f:
																			lowerCamelCase__     :					Tuple		=f.read()

										if isinstance(lowerCamelCase_					,							lowerCamelCase_	):
													lowerCamelCase__     :					List[str]		=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__     :					Tuple		=inputs.pop('raw'					,							lowerCamelCase_	)
													if _inputs is None:
																# Remove path which will not be used from `datasets`.
																inputs.pop('path'					,							lowerCamelCase_	)
																lowerCamelCase__     :					Union[str, Any]		=inputs.pop('array'					,							lowerCamelCase_	)
													lowerCamelCase__     :					str		=inputs.pop('sampling_rate'	)
													lowerCamelCase__     :					List[str]		=_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__     :					Optional[int]		=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__     :					str		=self.feature_extractor(
										    lowerCamelCase_					,							sampling_rate=self.feature_extractor.sampling_rate					,							return_tensors='pt'	)
										return processed





							def 				UpperCAmelCase__							(  self  :Dict					,							lowerCamelCase_  :Dict	):





										"""simple docstring"""

										lowerCamelCase__     :					int		=self.model(**lowerCamelCase_	)
										return model_outputs





							def 				UpperCAmelCase__							(  self  :List[str]					,							lowerCamelCase_  :int					,							lowerCamelCase_  :Dict=5	):





										"""simple docstring"""

										lowerCamelCase__     :					Optional[int]		=model_outputs.logits[0].softmax(-1	)
										lowerCamelCase__	,    lowerCamelCase__     :					str		=probs.topk(lowerCamelCase_	)

										lowerCamelCase__     :					Optional[int]		=scores.tolist()
										lowerCamelCase__     :					Any		=ids.tolist()

										lowerCamelCase__     :					Union[str, Any]		=[{'score': score, 'label': self.model.config.idalabel[_id]} for score, _id in zip(lowerCamelCase_					,							lowerCamelCase_	)]

										return labels

"""simple docstring"""


from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


lowerCAmelCase								=						{
    """configuration_deberta_v2""": ["""DEBERTA_V2_PRETRAINED_CONFIG_ARCHIVE_MAP""", """DebertaV2Config""", """DebertaV2OnnxConfig"""],
    """tokenization_deberta_v2""": ["""DebertaV2Tokenizer"""],
}

try:
							if not is_tokenizers_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							lowerCAmelCase								=						["""DebertaV2TokenizerFast"""]

try:
							if not is_tf_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							lowerCAmelCase								=						[
							    """TF_DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST""",
							    """TFDebertaV2ForMaskedLM""",
							    """TFDebertaV2ForQuestionAnswering""",
							    """TFDebertaV2ForSequenceClassification""",
							    """TFDebertaV2ForTokenClassification""",
							    """TFDebertaV2Model""",
							    """TFDebertaV2PreTrainedModel""",
							]

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							lowerCAmelCase								=						[
							    """DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST""",
							    """DebertaV2ForMaskedLM""",
							    """DebertaV2ForMultipleChoice""",
							    """DebertaV2ForQuestionAnswering""",
							    """DebertaV2ForSequenceClassification""",
							    """DebertaV2ForTokenClassification""",
							    """DebertaV2Model""",
							    """DebertaV2PreTrainedModel""",
							]


if TYPE_CHECKING:
							from .configuration_deberta_va import (
							    DEBERTA_V2_PRETRAINED_CONFIG_ARCHIVE_MAP,
							    DebertaVaConfig,
							    DebertaVaOnnxConfig,
							)
							from .tokenization_deberta_va import DebertaVaTokenizer

							try:
														if not is_tokenizers_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .tokenization_deberta_va_fast import DebertaVaTokenizerFast

							try:
														if not is_tf_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_tf_deberta_va import (
														    TF_DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST,
														    TFDebertaVaForMaskedLM,
														    TFDebertaVaForQuestionAnswering,
														    TFDebertaVaForSequenceClassification,
														    TFDebertaVaForTokenClassification,
														    TFDebertaVaModel,
														    TFDebertaVaPreTrainedModel,
														)

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_deberta_va import (
														    DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST,
														    DebertaVaForMaskedLM,
														    DebertaVaForMultipleChoice,
														    DebertaVaForQuestionAnswering,
														    DebertaVaForSequenceClassification,
														    DebertaVaForTokenClassification,
														    DebertaVaModel,
														    DebertaVaPreTrainedModel,
														)

else:
							import sys

							lowerCAmelCase								=						_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)