





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_					:   List[Any] 							=						logging.get_logger(__name__)





@add_end_docstrings(__A  )
class      lowercase__    (							__A  ):







  def __init__(	self    ,						**_lowercase						):
       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    ,						_lowercase    ,						**_lowercase						):
       return super().__call__(_lowercase    ,						**_lowercase						)







  def 	UpperCAmelCase__		(	self    ,						**_lowercase						):
       lowerCAmelCase_					:       Any    =		{}
       if "candidate_labels" in kwargs:
            lowerCAmelCase_					:       Optional[int]    =		kwargs["""candidate_labels"""]
       if "hypothesis_template" in kwargs:
            lowerCAmelCase_					:       Union[str, Any]    =		kwargs["""hypothesis_template"""]

       return preprocess_params, {}, {}







  def 	UpperCAmelCase__		(	self    ,						_lowercase    ,						_lowercase=None    ,						_lowercase="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
                 lowerCAmelCase_					:       List[str]    =		requests.get(_lowercase						).content
            else:
                 with open(_lowercase    ,						"""rb"""						) as f:
                      lowerCAmelCase_					:       Any    =		f.read()

       if isinstance(_lowercase    ,						_lowercase						):
            lowerCAmelCase_					:       List[Any]    =		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"""						)

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







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):
       lowerCAmelCase_					:       Any    =		model_inputs.pop("""candidate_labels"""						)
       lowerCAmelCase_					:       Optional[Any]    =		model_inputs.pop("""text_inputs"""						)
       if isinstance(text_inputs[0]    ,						_lowercase						):
            lowerCAmelCase_					:       Optional[Any]    =		text_inputs[0]
       else:
            # Batching case.
            lowerCAmelCase_					:       Dict    =		text_inputs[0][0]

       lowerCAmelCase_					:       Any    =		self.model(**_lowercase    ,						**_lowercase						)

       lowerCAmelCase_					:       List[Any]    =		{
           """candidate_labels""": candidate_labels,
           """logits""": outputs.logits_per_audio,
       }
       return model_outputs







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):
       lowerCAmelCase_					:       int    =		model_outputs.pop("""candidate_labels"""						)
       lowerCAmelCase_					:       Any    =		model_outputs["""logits"""][0]

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

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







from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCAmelCase_					:   Tuple 							=						logging.get_logger(__name__)

UpperCAmelCase_					:   List[str] 							=						{
    """google/fnet-base""": """https://huggingface.co/google/fnet-base/resolve/main/config.json""",
    """google/fnet-large""": """https://huggingface.co/google/fnet-large/resolve/main/config.json"""
    # See all FNet models at https://huggingface.co/models?filter=fnet
}





class      lowercase__    (							__A  ):
  __UpperCamelCase	     =  """fnet"""







  def __init__(	self    ,						_lowercase=32_000    ,						_lowercase=768    ,						_lowercase=12    ,						_lowercase=3_072    ,						_lowercase="gelu_new"    ,						_lowercase=0.1    ,						_lowercase=512    ,						_lowercase=4    ,						_lowercase=0.02    ,						_lowercase=1e-12    ,						_lowercase=False    ,						_lowercase=512    ,						_lowercase=3    ,						_lowercase=1    ,						_lowercase=2    ,						**_lowercase    ,						):
       super().__init__(pad_token_id=_lowercase    ,						bos_token_id=_lowercase    ,						eos_token_id=_lowercase    ,						**_lowercase						)

       lowerCAmelCase_					:       Optional[Any]    =		vocab_size
       lowerCAmelCase_					:       List[str]    =		max_position_embeddings
       lowerCAmelCase_					:       str    =		hidden_size
       lowerCAmelCase_					:       Any    =		num_hidden_layers
       lowerCAmelCase_					:       Optional[int]    =		intermediate_size
       lowerCAmelCase_					:       int    =		hidden_act
       lowerCAmelCase_					:       List[str]    =		hidden_dropout_prob
       lowerCAmelCase_					:       Tuple    =		initializer_range
       lowerCAmelCase_					:       int    =		type_vocab_size
       lowerCAmelCase_					:       str    =		layer_norm_eps
       lowerCAmelCase_					:       Optional[Any]    =		use_tpu_fourier_optimizations
       lowerCAmelCase_					:       Union[str, Any]    =		tpu_short_seq_length
