



'''simple docstring'''
from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


snake_case_					:      Any    			=    logging.get_logger(__name__)






@add_end_docstrings(_a  )
class 				lowercase__      (				_a  ):





	def __init__(					self   :							Union[str, Any]					,**lowerCamelCase__   :							Any					):







							'''simple docstring'''
							super().__init__(**lowerCamelCase__					)

							if self.framework != "pt":
													raise ValueError(F'The {self.__class__} is only available in PyTorch.'					)
							# No specific FOR_XXX available yet





	def __call__(					self   :							Any					,lowerCamelCase__   :							Union[np.ndarray, bytes, str]					,**lowerCamelCase__   :							Optional[int]					):







							'''simple docstring'''
							return super().__call__(lowerCamelCase__					,**lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							Optional[Any]					,**lowerCamelCase__   :							Union[str, Any]					):







							'''simple docstring'''
							_UpperCamelCase   :						str												=  {}
							if "candidate_labels" in kwargs:
													_UpperCamelCase   :						Any												=  kwargs['candidate_labels']
							if "hypothesis_template" in kwargs:
													_UpperCamelCase   :						Optional[Any]												=  kwargs['hypothesis_template']

							return preprocess_params, {}, {}





	def       UpperCamelCase_    (					self   :							List[Any]					,lowerCamelCase__   :							Union[str, Any]					,lowerCamelCase__   :							Dict=None					,lowerCamelCase__   :							Dict="This is a sound of {}."					):







							'''simple docstring'''

							if isinstance(lowerCamelCase__					,lowerCamelCase__					):
													if audio.startswith('http://'					) or audio.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   :						Optional[int]												=  requests.get(lowerCamelCase__					).content
													else:
																			with open(lowerCamelCase__					,'rb'					) as f:
																									_UpperCamelCase   :						List[Any]												=  f.read()

							if isinstance(lowerCamelCase__					,lowerCamelCase__					):
													_UpperCamelCase   :						Union[str, Any]												=  ffmpeg_read(lowerCamelCase__					,self.feature_extractor.sampling_rate					)

							if not isinstance(lowerCamelCase__					,np.ndarray					):
													raise ValueError('We expect a numpy ndarray as input'					)
							if len(audio.shape					) != 1:
													raise ValueError('We expect a single channel audio input for ZeroShotAudioClassificationPipeline'					)

							_UpperCamelCase   :						List[Any]												=  self.feature_extractor(
							    [audio]					,sampling_rate=self.feature_extractor.sampling_rate					,return_tensors='pt'					)
							_UpperCamelCase   :						Optional[int]												=  candidate_labels
							_UpperCamelCase   :						int												=  [hypothesis_template.format(lowerCamelCase__					) for x in candidate_labels]
							_UpperCamelCase   :						Optional[int]												=  self.tokenizer(lowerCamelCase__					,return_tensors=self.framework					,padding=lowerCamelCase__					)
							_UpperCamelCase   :						List[str]												=  [text_inputs]
							return inputs





	def       UpperCamelCase_    (					self   :							int					,lowerCamelCase__   :							Union[str, Any]					):







							'''simple docstring'''
							_UpperCamelCase   :						Dict												=  model_inputs.pop('candidate_labels'					)
							_UpperCamelCase   :						Dict												=  model_inputs.pop('text_inputs'					)
							if isinstance(text_inputs[0]					,lowerCamelCase__					):
													_UpperCamelCase   :						Union[str, Any]												=  text_inputs[0]
							else:
													# Batching case.
													_UpperCamelCase   :						Tuple												=  text_inputs[0][0]

							_UpperCamelCase   :						Tuple												=  self.model(**lowerCamelCase__					,**lowerCamelCase__					)

							_UpperCamelCase   :						Optional[Any]												=  {
							    'candidate_labels': candidate_labels,
							    'logits': outputs.logits_per_audio,
							}
							return model_outputs





	def       UpperCamelCase_    (					self   :							List[str]					,lowerCamelCase__   :							Dict					):







							'''simple docstring'''
							_UpperCamelCase   :						List[str]												=  model_outputs.pop('candidate_labels'					)
							_UpperCamelCase   :						Dict												=  model_outputs['logits'][0]

							if self.framework == "pt":
													_UpperCamelCase   :						Dict												=  logits.softmax(dim=0					)
													_UpperCamelCase   :						Optional[Any]												=  probs.tolist()
							else:
													raise ValueError('`tf` framework not supported.'					)

							_UpperCamelCase   :						Union[str, Any]												=  [
							    {'score': score, 'label': candidate_label}
							    for score, candidate_label in sorted(zip(lowerCamelCase__					,lowerCamelCase__					)					,key=lambda lowerCamelCase__					: -x[0]					)
							]
							return result






# flake8: noqa
# Lint as: python3

from typing import Dict, List, Optional, Type

from .. import config
from ..utils import logging
from .formatting import (
    ArrowFormatter,
    CustomFormatter,
    Formatter,
    PandasFormatter,
    PythonFormatter,
    TensorFormatter,
    format_table,
    query_table,
)
from .np_formatter import NumpyFormatter


A				:       Union[str, Any]      =  logging.get_logger(__name__)

A				:       Dict[Optional[str], Type[Formatter]]      =  {}
A				:       Dict[Optional[str], str]      =  {}
A				:       Dict[Optional[str], Exception]      =  {}





def 						_lowerCAmelCase							(     _lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase = None   ,						)    ->						str:


						'''simple docstring'''






						__snake_case				=		aliases if aliases is not None else []
						if format_type in _FORMAT_TYPES:
												logger.warning(
												    F'''Overwriting format type \'{format_type}\' ({_FORMAT_TYPES[format_type].__name__} -> {formatter_cls.__name__})'''			)
						__snake_case				=		formatter_cls
						for alias in set(aliases + [format_type]			):
												if alias in _FORMAT_TYPES_ALIASES:
																		logger.warning(
																		    F'''Overwriting format type alias \'{alias}\' ({_FORMAT_TYPES_ALIASES[alias]} -> {format_type})'''			)
												__snake_case				=		format_type





def 						_lowerCAmelCase							(     _lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase = None			)    ->						List[Any]:


						'''simple docstring'''






						__snake_case				=		aliases if aliases is not None else []
						for alias in set(aliases + [format_type]			):
												__snake_case				=		unavailable_error


# Here we define all the available formatting functions that can be used by `Dataset.set_format`
_register_formatter(PythonFormatter, None, aliases=['python'])
_register_formatter(ArrowFormatter, 'arrow', aliases=['pa', 'pyarrow'])
_register_formatter(NumpyFormatter, 'numpy', aliases=['np'])
_register_formatter(PandasFormatter, 'pandas', aliases=['pd'])
_register_formatter(CustomFormatter, 'custom')

if config.TORCH_AVAILABLE:
	from .torch_formatter import TorchFormatter

	_register_formatter(TorchFormatter, 'torch', aliases=['pt', 'pytorch'])
else:
	A				:       Any      =  ValueError('PyTorch needs to be installed to be able to return PyTorch tensors.')
	_register_unavailable_formatter(_torch_error, 'torch', aliases=['pt', 'pytorch'])

if config.TF_AVAILABLE:
	from .tf_formatter import TFFormatter

	_register_formatter(TFFormatter, 'tensorflow', aliases=['tf'])
else:
	A				:       Tuple      =  ValueError('Tensorflow needs to be installed to be able to return Tensorflow tensors.')
	_register_unavailable_formatter(_tf_error, 'tensorflow', aliases=['tf'])

if config.JAX_AVAILABLE:
	from .jax_formatter import JaxFormatter

	_register_formatter(JaxFormatter, 'jax', aliases=[])
else:
	A				:       Optional[int]      =  ValueError('JAX needs to be installed to be able to return JAX arrays.')
	_register_unavailable_formatter(_jax_error, 'jax', aliases=[])





def 						_lowerCAmelCase							(     _lowerCAmelCase			)    ->						Optional[str]:


						'''simple docstring'''






						if format_type in _FORMAT_TYPES_ALIASES:
												return _FORMAT_TYPES_ALIASES[format_type]
						else:
												return format_type





def 						_lowerCAmelCase							(     _lowerCAmelCase   ,						**_lowerCAmelCase			)    ->						Formatter:


						'''simple docstring'''






						__snake_case				=		get_format_type_from_alias(_lowerCAmelCase			)
						if format_type in _FORMAT_TYPES:
												return _FORMAT_TYPES[format_type](**_lowerCAmelCase			)
						if format_type in _FORMAT_TYPES_ALIASES_UNAVAILABLE:
												raise _FORMAT_TYPES_ALIASES_UNAVAILABLE[format_type]
						else:
												raise ValueError(
												    F'''Return type should be None or selected in {list(type for type in _FORMAT_TYPES.keys() if type != None			)}, but got \'{format_type}\''''			)





