





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


_UpperCAmelCase       =				logging.get_logger(__name__)






@add_end_docstrings(_A		)
class   snake_case_							(						_A		):






  def __init__(				self					:	Dict  ,    **_snake_case					:	Union[str, Any]     )->Dict:






   '''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					:	Tuple  ,    _snake_case					:	Union[np.ndarray, bytes, str]  ,    **_snake_case					:	Union[str, Any]     )->Dict:






   '''simple docstring'''

   return super().__call__(__lowerCAmelCase  ,    **__lowerCAmelCase     )






  def  UpperCAmelCase__		(				self					:	Optional[Any]  ,    **_snake_case					:	Optional[int]     )->Optional[int]:






   '''simple docstring'''

   __lowerCAmelCase	:			Optional[int]         =  {}
   if "candidate_labels" in kwargs:
    __lowerCAmelCase	:			int         =  kwargs["""candidate_labels"""]
   if "hypothesis_template" in kwargs:
    __lowerCAmelCase	:			int         =  kwargs["""hypothesis_template"""]

   return preprocess_params, {}, {}






  def  UpperCAmelCase__		(				self					:	str  ,    _snake_case					:	Union[str, Any]  ,    _snake_case					:	Dict=None  ,    _snake_case					:	str="This is a sound of {}."     )->Optional[Any]:






   '''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
     __lowerCAmelCase	:			str         =  requests.get(__lowerCAmelCase     ).content
    else:
     with open(__lowerCAmelCase  ,    """rb"""     ) as f:
      __lowerCAmelCase	:			int         =  f.read()

   if isinstance(__lowerCAmelCase  ,    __lowerCAmelCase     ):
    __lowerCAmelCase	:			str         =  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"""     )

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






  def  UpperCAmelCase__		(				self					:	List[str]  ,    _snake_case					:	Union[str, Any]     )->Dict:






   '''simple docstring'''

   __lowerCAmelCase	:			Optional[int]         =  model_inputs.pop("""candidate_labels"""     )
   __lowerCAmelCase	:			List[Any]         =  model_inputs.pop("""text_inputs"""     )
   if isinstance(text_inputs[0]  ,    __lowerCAmelCase     ):
    __lowerCAmelCase	:			Any         =  text_inputs[0]
   else:
    # Batching case.
    __lowerCAmelCase	:			Any         =  text_inputs[0][0]

   __lowerCAmelCase	:			Dict         =  self.model(**__lowerCAmelCase  ,    **__lowerCAmelCase     )

   __lowerCAmelCase	:			Union[str, Any]         =  {
       """candidate_labels""": candidate_labels,
       """logits""": outputs.logits_per_audio,
   }
   return model_outputs






  def  UpperCAmelCase__		(				self					:	Tuple  ,    _snake_case					:	List[str]     )->Tuple:






   '''simple docstring'''

   __lowerCAmelCase	:			int         =  model_outputs.pop("""candidate_labels"""     )
   __lowerCAmelCase	:			List[str]         =  model_outputs["""logits"""][0]

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

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




def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:list				)  ->     bool:
   if not isinstance(_snake_case	,   _snake_case				):
      raise ValueError('''Input series is not valid, valid series - [2, 4, 6]'''				)
   if len(_snake_case				) == 0:
      raise ValueError('''Input list must be a non empty list'''				)
   if len(_snake_case				) == 1:
      return True
   _A      =       series[1] - series[0]
   for index in range(len(_snake_case				) - 1				):
      if series[index + 1] - series[index] != common_diff:
         return False
   return True

def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:list				)  ->     float:
   if not isinstance(_snake_case	,   _snake_case				):
      raise ValueError('''Input series is not valid, valid series - [2, 4, 6]'''				)
   if len(_snake_case				) == 0:
      raise ValueError('''Input list must be a non empty list'''				)
   _A      =       0
   for val in series:
      answer += val
   return answer / len(_snake_case				)


if __name__ == "__main__":
   import doctest

   doctest.testmod()



