




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


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






@add_end_docstrings(_lowerCamelCase     )
class 						a				(   _lowerCamelCase     ):


				def __init__(		self    :					Any       ,				**lowercase_    :					Any					):
									super().__init__(**lowercase_					)

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


				def __call__(		self    :					Dict       ,				lowercase_    :					Union[np.ndarray, bytes, str]       ,				**lowercase_    :					Dict					):
									return super().__call__(lowercase_       ,				**lowercase_					)


				def   A_ (		self    :					Union[str, Any]       ,				**lowercase_    :					List[str]					):
									snake_case_									=				{}
									if "candidate_labels" in kwargs:
														snake_case_									=				kwargs['''candidate_labels''']
									if "hypothesis_template" in kwargs:
														snake_case_									=				kwargs['''hypothesis_template''']

									return preprocess_params, {}, {}


				def   A_ (		self    :					str       ,				lowercase_    :					List[str]       ,				lowercase_    :					Optional[int]=None       ,				lowercase_    :					Tuple="This is a sound of {}."					):
									if isinstance(lowercase_       ,				lowercase_					):
														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
																			snake_case_									=				requests.get(lowercase_					).content
														else:
																			with open(lowercase_       ,				'''rb'''					) as f:
																								snake_case_									=				f.read()

									if isinstance(lowercase_       ,				lowercase_					):
														snake_case_									=				ffmpeg_read(lowercase_       ,				self.feature_extractor.sampling_rate					)

									if not isinstance(lowercase_       ,				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'''					)

									snake_case_									=				self.feature_extractor(
									    [audio]       ,				sampling_rate=self.feature_extractor.sampling_rate       ,				return_tensors='''pt'''					)
									snake_case_									=				candidate_labels
									snake_case_									=				[hypothesis_template.format(lowercase_					) for x in candidate_labels]
									snake_case_									=				self.tokenizer(lowercase_       ,				return_tensors=self.framework       ,				padding=lowercase_					)
									snake_case_									=				[text_inputs]
									return inputs


				def   A_ (		self    :					int       ,				lowercase_    :					str					):
									snake_case_									=				model_inputs.pop('''candidate_labels'''					)
									snake_case_									=				model_inputs.pop('''text_inputs'''					)
									if isinstance(text_inputs[0]       ,				lowercase_					):
														snake_case_									=				text_inputs[0]
									else:
														# Batching case.
														snake_case_									=				text_inputs[0][0]

									snake_case_									=				self.model(**lowercase_       ,				**lowercase_					)

									snake_case_									=				{
									    '''candidate_labels''': candidate_labels,
									    '''logits''': outputs.logits_per_audio,
									}
									return model_outputs


				def   A_ (		self    :					Any       ,				lowercase_    :					int					):
									snake_case_									=				model_outputs.pop('''candidate_labels'''					)
									snake_case_									=				model_outputs['''logits'''][0]

									if self.framework == "pt":
														snake_case_									=				logits.softmax(dim=0					)
														snake_case_									=				probs.tolist()
									else:
														raise ValueError('''`tf` framework not supported.'''					)

									snake_case_									=				[
									    {'''score''': score, '''label''': candidate_label}
									    for score, candidate_label in sorted(zip(lowercase_       ,				lowercase_					)       ,				key=lambda lowercase_					: -x[0]					)
									]
									return result








"""simple docstring"""
from typing import TYPE_CHECKING

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


_snake_case     :		Union[str, Any]       						= {
    'configuration_encodec': [
        'ENCODEC_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'EncodecConfig',
    ],
    'feature_extraction_encodec': ['EncodecFeatureExtractor'],
}

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_snake_case     :		int       						= [
		    'ENCODEC_PRETRAINED_MODEL_ARCHIVE_LIST',
		    'EncodecModel',
		    'EncodecPreTrainedModel',
		]

if TYPE_CHECKING:
		from .configuration_encodec import (
		    ENCODEC_PRETRAINED_CONFIG_ARCHIVE_MAP,
		    EncodecConfig,
		)
		from .feature_extraction_encodec import EncodecFeatureExtractor

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_encodec import (
				    ENCODEC_PRETRAINED_MODEL_ARCHIVE_LIST,
				    EncodecModel,
				    EncodecPreTrainedModel,
				)

else:
		import sys

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


