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


lowercase        =  logging.get_logger(__name__)





@add_end_docstrings(UpperCAmelCase       )
class       __A(				UpperCAmelCase       ):

					def __init__(	self					:     int   ,					**__UpperCamelCase					:     Optional[Any]   ):
												super().__init__(**__UpperCamelCase   )

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

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

					def    lowercase__      (	self					:     Union[str, Any]   ,					**__UpperCamelCase					:     Any   ):
												lowerCamelCase_							  =					{}
												if "candidate_labels" in kwargs:
																			lowerCamelCase_							  =					kwargs["""candidate_labels"""]
												if "hypothesis_template" in kwargs:
																			lowerCamelCase_							  =					kwargs["""hypothesis_template"""]

												return preprocess_params, {}, {}

					def    lowercase__      (	self					:     List[Any]   ,					__UpperCamelCase					:     Optional[int]   ,					__UpperCamelCase					:     str=None   ,					__UpperCamelCase					:     List[str]="This is a sound of {}."   ):

												if isinstance(__UpperCamelCase   ,					__UpperCamelCase   ):
																			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
																										lowerCamelCase_							  =					requests.get(__UpperCamelCase   ).content
																			else:
																										with open(__UpperCamelCase   ,					"""rb"""   ) as f:
																																	lowerCamelCase_							  =					f.read()

												if isinstance(__UpperCamelCase   ,					__UpperCamelCase   ):
																			lowerCamelCase_							  =					ffmpeg_read(__UpperCamelCase   ,					self.feature_extractor.sampling_rate   )

												if not isinstance(__UpperCamelCase   ,					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"""   )

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

					def    lowercase__      (	self					:     Optional[int]   ,					__UpperCamelCase					:     Dict   ):
												lowerCamelCase_							  =					model_inputs.pop("""candidate_labels"""   )
												lowerCamelCase_							  =					model_inputs.pop("""text_inputs"""   )
												if isinstance(text_inputs[0]   ,					__UpperCamelCase   ):
																			lowerCamelCase_							  =					text_inputs[0]
												else:
																			# Batching case.
																			lowerCamelCase_							  =					text_inputs[0][0]

												lowerCamelCase_							  =					self.model(**__UpperCamelCase   ,					**__UpperCamelCase   )

												lowerCamelCase_							  =					{
												    """candidate_labels""": candidate_labels,
												    """logits""": outputs.logits_per_audio,
												}
												return model_outputs



					def    lowercase__      (	self					:     Dict   ,					__UpperCamelCase					:     str   ):
												lowerCamelCase_							  =					model_outputs.pop("""candidate_labels"""   )
												lowerCamelCase_							  =					model_outputs["""logits"""][0]

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

												lowerCamelCase_							  =					[
												    {"""score""": score, """label""": candidate_label}
												    for score, candidate_label in sorted(zip(__UpperCamelCase   ,					__UpperCamelCase   )   ,					key=lambda __UpperCamelCase   : -x[0]   )
												]
												return result







from ..utils import DummyObject, requires_backends





class       __A(				metaclass=UpperCAmelCase       ):
					SCREAMING_SNAKE_CASE			    =    ['''torch''', '''scipy''']

					def __init__(	self					:     Any   ,					*__UpperCamelCase					:     str   ,					**__UpperCamelCase					:     Any   ):
												requires_backends(self   ,					["""torch""", """scipy"""]   )

					@classmethod
					def    lowercase__      (	cls					:     Dict   ,					*__UpperCamelCase					:     str   ,					**__UpperCamelCase					:     Dict   ):
												requires_backends(cls   ,					["""torch""", """scipy"""]   )



					@classmethod
					def    lowercase__      (	cls					:     str   ,					*__UpperCamelCase					:     Optional[int]   ,					**__UpperCamelCase					:     Optional[int]   ):
												requires_backends(cls   ,					["""torch""", """scipy"""]   )





