



'''simple docstring'''






from typing import TYPE_CHECKING

from ...utils import _LazyModule


lowerCAmelCase:      Optional[int]		 =		{'processing_wav2vec2_with_lm': ['Wav2Vec2ProcessorWithLM']}


if TYPE_CHECKING:
     from .processing_wavaveca_with_lm import WavaVecaProcessorWithLM
else:
     import sys

     lowerCAmelCase:      Union[str, Any]		 =		_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)




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


lowerCAmelCase:      Optional[Any]		 =		logging.get_logger(__name__)

@add_end_docstrings(lowerCamelCase__			)
class 							a__(      lowerCamelCase__			):







    def __init__(   self				:						Union[str, Any]		,    **__snake_case				:						List[str]			):
      super().__init__(**__snake_case			)

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







    def __call__(   self				:						Any		,    __snake_case				:						Union[np.ndarray, bytes, str]		,    **__snake_case				:						Optional[Any]			):
      return super().__call__(__snake_case		,    **__snake_case			)







    def     lowercase_ (   self				:						Dict		,    **__snake_case				:						Optional[int]			):
      a		:      str       =  {}
      if "candidate_labels" in kwargs:
        a		:      Optional[int]       =  kwargs['candidate_labels']
      if "hypothesis_template" in kwargs:
        a		:      Any       =  kwargs['hypothesis_template']

      return preprocess_params, {}, {}







    def     lowercase_ (   self				:						int		,    __snake_case				:						Optional[Any]		,    __snake_case				:						List[Any]=None		,    __snake_case				:						List[str]="This is a sound of {}."			):
      if isinstance(__snake_case		,    __snake_case			):
        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
          a		:      List[Any]       =  requests.get(__snake_case			).content
        else:
          with open(__snake_case		,    'rb'			) as f:
            a		:      List[str]       =  f.read()

      if isinstance(__snake_case		,    __snake_case			):
        a		:      Optional[Any]       =  ffmpeg_read(__snake_case		,    self.feature_extractor.sampling_rate			)

      if not isinstance(__snake_case		,    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'			)

      a		:      str       =  self.feature_extractor(
          [audio]		,    sampling_rate=self.feature_extractor.sampling_rate		,    return_tensors='pt'			)
      a		:      Tuple       =  candidate_labels
      a		:      Union[str, Any]       =  [hypothesis_template.format(__snake_case			) for x in candidate_labels]
      a		:      int       =  self.tokenizer(__snake_case		,    return_tensors=self.framework		,    padding=__snake_case			)
      a		:      Any       =  [text_inputs]
      return inputs







    def     lowercase_ (   self				:						Optional[Any]		,    __snake_case				:						str			):
      a		:      Any       =  model_inputs.pop('candidate_labels'			)
      a		:      List[Any]       =  model_inputs.pop('text_inputs'			)
      if isinstance(text_inputs[0]		,    __snake_case			):
        a		:      str       =  text_inputs[0]
      else:
        # Batching case.
        a		:      Dict       =  text_inputs[0][0]

      a		:      List[Any]       =  self.model(**__snake_case		,    **__snake_case			)

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







    def     lowercase_ (   self				:						Union[str, Any]		,    __snake_case				:						str			):
      a		:      Optional[int]       =  model_outputs.pop('candidate_labels'			)
      a		:      List[str]       =  model_outputs['logits'][0]

      if self.framework == "pt":
        a		:      str       =  logits.softmax(dim=0			)
        a		:      List[Any]       =  probs.tolist()
      else:
        raise ValueError('`tf` framework not supported.'			)

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