


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			=						logging.get_logger(__name__)






@add_end_docstrings(A_		)
class    UpperCAmelCase__	( A_		):



				"""simple docstring"""






				def __init__(      self ,       **A_   )  ->	Dict:
									super().__init__(**A_   )

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






				def __call__(      self ,       A_ ,       **A_   )  ->	Union[str, Any]:
									return super().__call__(A_ ,       **A_   )






				def _a (      self ,       **A_   )  ->	List[Any]:
									__UpperCamelCase   							={}
									if "candidate_labels" in kwargs:
														__UpperCamelCase   							=kwargs['candidate_labels']
									if "hypothesis_template" in kwargs:
														__UpperCamelCase   							=kwargs['hypothesis_template']

									return preprocess_params, {}, {}






				def _a (      self ,       A_ ,       A_=None ,       A_="This is a sound of {}."   )  ->	Any:
									if isinstance(A_ ,       A_   ):
														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   							=requests.get(A_   ).content
														else:
																			with open(A_ ,       'rb'   ) as f:
																								__UpperCamelCase   							=f.read()

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

									if not isinstance(A_ ,       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   							=self.feature_extractor(
									    [audio] ,       sampling_rate=self.feature_extractor.sampling_rate ,       return_tensors='pt'   )
									__UpperCamelCase   							=candidate_labels
									__UpperCamelCase   							=[hypothesis_template.format(A_   ) for x in candidate_labels]
									__UpperCamelCase   							=self.tokenizer(A_ ,       return_tensors=self.framework ,       padding=A_   )
									__UpperCamelCase   							=[text_inputs]
									return inputs






				def _a (      self ,       A_   )  ->	int:
									__UpperCamelCase   							=model_inputs.pop('candidate_labels'   )
									__UpperCamelCase   							=model_inputs.pop('text_inputs'   )
									if isinstance(text_inputs[0] ,       A_   ):
														__UpperCamelCase   							=text_inputs[0]
									else:
														# Batching case.
														__UpperCamelCase   							=text_inputs[0][0]

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

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






				def _a (      self ,       A_   )  ->	Any:
									__UpperCamelCase   							=model_outputs.pop('candidate_labels'   )
									__UpperCamelCase   							=model_outputs['logits'][0]

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

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









import sys

from .dependency_versions_table import deps
from .utils.versions import require_version, require_version_core


# define which module versions we always want to check at run time
# (usually the ones defined in `install_requires` in setup.py)
#
# order specific notes:
# - tqdm must be checked before tokenizers

_A			=						'python tqdm regex requests packaging filelock numpy tokenizers'.split()
if sys.version_info < (3, 7):
   pkgs_to_check_at_runtime.append('dataclasses')
if sys.version_info < (3, 8):
   pkgs_to_check_at_runtime.append('importlib_metadata')

for pkg in pkgs_to_check_at_runtime:
   if pkg in deps:
      if pkg == "tokenizers":
         # must be loaded here, or else tqdm check may fail
         from .utils import is_tokenizers_available

         if not is_tokenizers_available():
            continue  # not required, check version only if installed

      require_version_core(deps[pkg])
   else:
      raise ValueError(f"""can't find {pkg} in {deps.keys()}, check dependency_versions_table.py""")

def 						_UpperCAmelCase    (      SCREAMING_SNAKE_CASE__      : Union[str, Any]      ,					SCREAMING_SNAKE_CASE__      : Optional[int]=None    ):
     require_version(deps[pkg]      ,					SCREAMING_SNAKE_CASE__    )





