
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 os

from apache_beam.io.filesystems import FileSystems
from apache_beam.pipeline import Pipeline

from .logging import get_logger


A  :			Optional[int]     				=				2 << 2_0  # 2mb
A  :			Optional[int]     				=				get_logger(__name__)







class     __A(				a    ):




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


								'''simple docstring'''




								__a    							=					self._options.get_all_options().get('''runner'''       )
								return runner in [None, "DirectRunner", "PortableRunner"]



def 					__lowerCAmelCase    (    a__		,					a__		,					a__=False						) ->						Union[str, Any]:
				__a    							=					FileSystems
				if fs.exists(a__						):
								if force_upload:
												logger.info(F"""Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True."""						)
								else:
												logger.info(F"""Remote path already exist: {remote_file_path}. Skipping it as force_upload=False."""						)
												return
				with fs.create(a__						) as remote_file:
								with open(a__		,					'''rb'''						) as local_file:
												__a    							=					local_file.read(a__						)
												while chunk:
																remote_file.write(a__						)
																__a    							=					local_file.read(a__						)






def 					__lowerCAmelCase    (    a__		,					a__		,					a__=False						) ->						Dict:
				__a    							=					FileSystems
				if os.path.exists(a__						):
								if force_download:
												logger.info(F"""Local path already exist: {remote_file_path}. Overwriting it as force_upload=True."""						)
								else:
												logger.info(F"""Local path already exist: {remote_file_path}. Skipping it as force_upload=False."""						)
												return
				with fs.open(a__						) as remote_file:
								with open(a__		,					'''wb'''						) as local_file:
												__a    							=					remote_file.read(a__						)
												while chunk:
																local_file.write(a__						)
																__a    							=					remote_file.read(a__						)