


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


lowerCAmelCase__  :Optional[int]             =							logging.get_logger(__name__)

@add_end_docstrings(UpperCAmelCase      )
class 		__a				(    UpperCAmelCase      ):
		def __init__(      self      ,   **_SCREAMING_SNAKE_CASE						)			->	str:




									"""simple docstring"""





									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      ,   _SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)			->	Union[str, Any]:




									"""simple docstring"""





									return super().__call__(__A      ,   **__A						)
		def      UpperCAmelCase__    (      self      ,   **_SCREAMING_SNAKE_CASE						)			->	str:




									"""simple docstring"""





									_UpperCAmelCase					=      {}
									if "candidate_labels" in kwargs:
																_UpperCAmelCase					=      kwargs['candidate_labels']
									if "hypothesis_template" in kwargs:
																_UpperCAmelCase					=      kwargs['hypothesis_template']

									return preprocess_params, {}, {}
		def      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE=None      ,   _SCREAMING_SNAKE_CASE="This is a sound of {}."						)			->	List[Any]:




									"""simple docstring"""





									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      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE						)			->	List[str]:




									"""simple docstring"""





									_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      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE						)			->	Any:




									"""simple docstring"""





									_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 _SCREAMING_SNAKE_CASE						: -x[0]						)
									]
									return result




from importlib import import_module

from .logging import get_logger


lowerCAmelCase__  :Optional[Any]             =							get_logger(__name__)

class 		__a				:
		def __init__(      self      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE=None						)			->	List[Any]:




									"""simple docstring"""





									_UpperCAmelCase					=      attrs or []
									if module is not None:
																for key in module.__dict__:
																							if key in attrs or not key.startswith('__'						):
																														setattr(self      ,   _SCREAMING_SNAKE_CASE      ,   getattr(_SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE						)						)
									_UpperCAmelCase					=      module._original_module if isinstance(_SCREAMING_SNAKE_CASE      ,   _PatchedModuleObj						) else module



class 		__a				:
		_a     :  Any           =     []
		def __init__(      self      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE=None						)			->	Union[str, Any]:




									"""simple docstring"""





									_UpperCAmelCase					=      obj
									_UpperCAmelCase					=      target
									_UpperCAmelCase					=      new
									_UpperCAmelCase					=      target.split('.'						)[0]
									_UpperCAmelCase					=      {}
									_UpperCAmelCase					=      attrs or []
		def __enter__(      self						)			->	int:




									"""simple docstring"""





									*_UpperCAmelCase		,	_UpperCAmelCase					=      self.target.split('.'						)

									# Patch modules:
									# it's used to patch attributes of submodules like "os.path.join";
									# in this case we need to patch "os" and "os.path"

									for i in range(len(_SCREAMING_SNAKE_CASE						)						):
																try:
																							_UpperCAmelCase					=      import_module('.'.join(submodules[: i + 1]						)						)
																except ModuleNotFoundError:
																							continue
																# We iterate over all the globals in self.obj in case we find "os" or "os.path"
																for attr in self.obj.__dir__():
																							_UpperCAmelCase					=      getattr(self.obj      ,   _SCREAMING_SNAKE_CASE						)
																							# We don't check for the name of the global, but rather if its value *is* "os" or "os.path".
																							# This allows to patch renamed modules like "from os import path as ospath".
																							if obj_attr is submodule or (
																							    (isinstance(_SCREAMING_SNAKE_CASE      ,   _PatchedModuleObj						) and obj_attr._original_module is submodule)
																							):
																														_UpperCAmelCase					=      obj_attr
																														# patch at top level
																														setattr(self.obj      ,   _SCREAMING_SNAKE_CASE      ,   _PatchedModuleObj(_SCREAMING_SNAKE_CASE      ,   attrs=self.attrs						)						)
																														_UpperCAmelCase					=      getattr(self.obj      ,   _SCREAMING_SNAKE_CASE						)
																														# construct lower levels patches
																														for key in submodules[i + 1 :]:
																																					setattr(_SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE      ,   _PatchedModuleObj(getattr(_SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE						)      ,   attrs=self.attrs						)						)
																																					_UpperCAmelCase					=      getattr(_SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE						)
																														# finally set the target attribute
																														setattr(_SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE      ,   self.new						)

        # Patch attribute itself:
        # it's used for builtins like "open",
        # and also to patch "os.path.join" we may also need to patch "join"
        # itself if it was imported as "from os.path import join".

									if submodules:  # if it's an attribute of a submodule like "os.path.join"
																try:
																							_UpperCAmelCase					=      getattr(import_module('.'.join(_SCREAMING_SNAKE_CASE						)						)      ,   _SCREAMING_SNAKE_CASE						)
																except (AttributeError, ModuleNotFoundError):
																							return
																# We iterate over all the globals in self.obj in case we find "os.path.join"
																for attr in self.obj.__dir__():
																							# We don't check for the name of the global, but rather if its value *is* "os.path.join".
																							# This allows to patch renamed attributes like "from os.path import join as pjoin".
																							if getattr(self.obj      ,   _SCREAMING_SNAKE_CASE						) is attr_value:
																														_UpperCAmelCase					=      getattr(self.obj      ,   _SCREAMING_SNAKE_CASE						)
																														setattr(self.obj      ,   _SCREAMING_SNAKE_CASE      ,   self.new						)
									elif target_attr in globals()["__builtins__"]:  # if it'a s builtin like "open"
																_UpperCAmelCase					=      globals()['__builtins__'][target_attr]
																setattr(self.obj      ,   _SCREAMING_SNAKE_CASE      ,   self.new						)
									else:
																raise RuntimeError(f'''Tried to patch attribute {target_attr} instead of a submodule.'''						)
		def __exit__(      self      ,   *_SCREAMING_SNAKE_CASE						)			->	Union[str, Any]:




									"""simple docstring"""





									for attr in list(self.original						):
																setattr(self.obj      ,   _SCREAMING_SNAKE_CASE      ,   self.original.pop(_SCREAMING_SNAKE_CASE						)						)
		def      UpperCAmelCase__    (      self						)			->	Optional[int]:




									"""simple docstring"""





									self.__enter__()
									self._active_patches.append(self						)






		def      UpperCAmelCase__    (      self						)			->	str:




									"""simple docstring"""





									try:
																self._active_patches.remove(self						)
									except ValueError:
																# If the patch hasn't been started this will fail
																return None

									return self.__exit__()
