





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

UpperCamelCase_	 =logging.get_logger(__name__)





def 			a_							(				_lowercase			,		_lowercase  ):
						_UpperCamelCase     :   Optional[int]      =       F"""{sampling_rate}"""
						_UpperCamelCase     :   int      =       '''1'''
						_UpperCamelCase     :   List[str]      =       '''f32le'''
						_UpperCamelCase     :   Optional[int]      =       [
						    '''ffmpeg''',
						    '''-i''',
						    '''pipe:0''',
						    '''-ac''',
						    ac,
						    '''-ar''',
						    ar,
						    '''-f''',
						    format_for_conversion,
						    '''-hide_banner''',
						    '''-loglevel''',
						    '''quiet''',
						    '''pipe:1''',
						]

						try:
												_UpperCamelCase     :   Dict      =       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'''  )
						_UpperCamelCase     :   Tuple      =       ffmpeg_process.communicate(_lowercase  )
						_UpperCamelCase     :   int      =       output_stream[0]

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






@add_end_docstrings(_lowerCAmelCase							)
class 					_a		( _lowerCAmelCase							):






				def __init__(					self						:       Tuple,					*lowerCAmelCase__						:       int,					**lowerCAmelCase__						:       List[Any]				)      -> Dict:




										'''simple docstring'''



										_UpperCamelCase     :   List[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						:       Dict,					lowerCAmelCase__						:       Union[np.ndarray, bytes, str],					**lowerCAmelCase__						:       List[Any],					)      -> Dict:




										'''simple docstring'''



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






				def 							snake_case      (					self						:       Dict,					lowerCAmelCase__						:       Dict=None,					**lowerCAmelCase__						:       Dict				)      -> str:




										'''simple docstring'''



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






				def 							snake_case      (					self						:       List[str],					lowerCAmelCase__						:       Optional[Any]				)      -> Optional[int]:




										'''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
																						_UpperCamelCase     :   Dict      =       requests.get(lowerCAmelCase__				).content
																else:
																						with open(lowerCAmelCase__,					'''rb'''				) as f:
																												_UpperCamelCase     :   Optional[Any]      =       f.read()

										if isinstance(lowerCAmelCase__,					lowerCAmelCase__				):
																_UpperCamelCase     :   Tuple      =       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'''				)

																_UpperCamelCase     :   Optional[Any]      =       inputs.pop('''raw''',					lowerCAmelCase__				)
																if _inputs is None:
																						# Remove path which will not be used from `datasets`.
																						inputs.pop('''path''',					lowerCAmelCase__				)
																						_UpperCamelCase     :   str      =       inputs.pop('''array''',					lowerCAmelCase__				)
																_UpperCamelCase     :   str      =       inputs.pop('''sampling_rate'''				)
																_UpperCamelCase     :   Dict      =       _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`.'''				)

																						_UpperCamelCase     :   Any      =       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'''				)

										_UpperCamelCase     :   Tuple      =       self.feature_extractor(
										    lowerCAmelCase__,					sampling_rate=self.feature_extractor.sampling_rate,					return_tensors='''pt'''				)
										return processed






				def 							snake_case      (					self						:       str,					lowerCAmelCase__						:       Dict				)      -> Optional[int]:




										'''simple docstring'''



										_UpperCamelCase     :   Tuple      =       self.model(**lowerCAmelCase__				)
										return model_outputs






				def 							snake_case      (					self						:       Optional[int],					lowerCAmelCase__						:       List[str],					lowerCAmelCase__						:       Dict=5				)      -> int:




										'''simple docstring'''



										_UpperCamelCase     :   List[Any]      =       model_outputs.logits[0].softmax(-1				)
										_UpperCamelCase  ,				_UpperCamelCase     :   Any      =       probs.topk(lowerCAmelCase__				)

										_UpperCamelCase     :   List[Any]      =       scores.tolist()
										_UpperCamelCase     :   List[Any]      =       ids.tolist()

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

										return labels






from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
    is_vision_available,
)


__A	:   Optional[int]	=					{
    'configuration_mobilevitv2': [
        'MOBILEVITV2_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'MobileViTV2Config',
        'MobileViTV2OnnxConfig',
    ],
}


try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__A	:   Tuple	=					[
	    'MOBILEVITV2_PRETRAINED_MODEL_ARCHIVE_LIST',
	    'MobileViTV2ForImageClassification',
	    'MobileViTV2ForSemanticSegmentation',
	    'MobileViTV2Model',
	    'MobileViTV2PreTrainedModel',
	]

if TYPE_CHECKING:
	from .configuration_mobilevitva import (
	    MOBILEVITV2_PRETRAINED_CONFIG_ARCHIVE_MAP,
	    MobileViTVaConfig,
	    MobileViTVaOnnxConfig,
	)

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_mobilevitva import (
		    MOBILEVITV2_PRETRAINED_MODEL_ARCHIVE_LIST,
		    MobileViTVaForImageClassification,
		    MobileViTVaForSemanticSegmentation,
		    MobileViTVaModel,
		    MobileViTVaPreTrainedModel,
		)

else:
	import sys

	__A	:   List[str]	=					_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)
