"""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 	UpperCAmelCase       (						UpperCamelCase__							,     UpperCamelCase__					):






							"""simple docstring"""





							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(__A			)
class      UpperCamelCase__(      __A			):





						def __init__(							self		,*__UpperCAmelCase		,**__UpperCAmelCase  )       ->					Optional[int]:
													# Default, might be overriden by the model.config.
													A__						  =				5
													super().__init__(*__UpperCAmelCase		,**__UpperCAmelCase  )

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

													self.check_model_type(__UpperCAmelCase  )





						def __call__(							self		,__UpperCAmelCase		,**__UpperCAmelCase		,)       ->					List[Any]:
													return super().__call__(__UpperCAmelCase		,**__UpperCAmelCase  )





						def 					snake_case__						(							self		,__UpperCAmelCase=None		,**__UpperCAmelCase  )       ->					List[Any]:
													# No parameters on this pipeline right now
													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 					snake_case__						(							self		,__UpperCAmelCase  )       ->					Union[str, Any]:
													if isinstance(__UpperCAmelCase		,__UpperCAmelCase  ):
																				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(__UpperCAmelCase  ).content
																				else:
																											with open(__UpperCAmelCase		,'rb'  ) as f:
																																		A__						  =				f.read()

													if isinstance(__UpperCAmelCase		,__UpperCAmelCase  ):
																				A__						  =				ffmpeg_read(__UpperCAmelCase		,self.feature_extractor.sampling_rate  )

													if isinstance(__UpperCAmelCase		,__UpperCAmelCase  ):
																				# 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'		,__UpperCAmelCase  )
																				if _inputs is None:
																											# Remove path which will not be used from `datasets`.
																											inputs.pop('path'		,__UpperCAmelCase  )
																											A__						  =				inputs.pop('array'		,__UpperCAmelCase  )
																				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(__UpperCAmelCase  )		,__UpperCAmelCase		,self.feature_extractor.sampling_rate  ).numpy()

													if not isinstance(__UpperCAmelCase		,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(
													    __UpperCAmelCase		,sampling_rate=self.feature_extractor.sampling_rate		,return_tensors='pt'  )
													return processed





						def 					snake_case__						(							self		,__UpperCAmelCase  )       ->					Optional[int]:
													A__						  =				self.model(**__UpperCAmelCase  )
													return model_outputs





						def 					snake_case__						(							self		,__UpperCAmelCase		,__UpperCAmelCase=5  )       ->					str:
													A__						  =				model_outputs.logits[0].softmax(-1  )
													A__  ,	A__						  =				probs.topk(__UpperCAmelCase  )

													A__						  =				scores.tolist()
													A__						  =				ids.tolist()

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

													return labels






"""simple docstring"""




def 	UpperCAmelCase       (						UpperCamelCase__ = 100							,     UpperCamelCase__ = 100					):






							"""simple docstring"""





							return max(
							    sum(int(UpperCamelCase__					) for x in str(base**power					)					)
							    for base in range(UpperCamelCase__					)
							    for power in range(UpperCamelCase__					)					)


# Tests
if __name__ == "__main__":
				import doctest

				doctest.testmod()





