





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



@add_end_docstrings(_UpperCamelCase )
class   __SCREAMING_SNAKE_CASE     (    _UpperCamelCase ):







  '''simple docstring'''


  def __init__(		self :							int       , **__a :							Dict						)		->					Optional[Any]:
        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 :							List[Any]       , __a :							Union[np.ndarray, bytes, str]       , **__a :							str						)		->					str:
        return super().__call__(__a       , **__a						)

  def  __SCREAMING_SNAKE_CASE					(		self :							Optional[Any]       , **__a :							int						)		->					str:
        _UpperCamelCase   :						Optional[int]    =							{}
        if "candidate_labels" in kwargs:
              _UpperCamelCase   :						Dict    =							kwargs["candidate_labels"]
        if "hypothesis_template" in kwargs:
              _UpperCamelCase   :						str    =							kwargs["hypothesis_template"]

        return preprocess_params, {}, {}

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

        if isinstance(__a       , __a						):
              _UpperCamelCase   :						List[str]    =							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   :						str    =							self.feature_extractor(
            [audio]       , sampling_rate=self.feature_extractor.sampling_rate       , return_tensors="pt"						)
        _UpperCamelCase   :						Any    =							candidate_labels
        _UpperCamelCase   :						Union[str, Any]    =							[hypothesis_template.format(__a						) for x in candidate_labels]
        _UpperCamelCase   :						Tuple    =							self.tokenizer(__a       , return_tensors=self.framework       , padding=__a						)
        _UpperCamelCase   :						Optional[int]    =							[text_inputs]
        return inputs

  def  __SCREAMING_SNAKE_CASE					(		self :							Any       , __a :							int						)		->					int:
        _UpperCamelCase   :						Dict    =							model_inputs.pop("candidate_labels"						)
        _UpperCamelCase   :						List[str]    =							model_inputs.pop("text_inputs"						)
        if isinstance(text_inputs[0]       , __a						):
              _UpperCamelCase   :						Dict    =							text_inputs[0]
        else:
              # Batching case.
              _UpperCamelCase   :						Union[str, Any]    =							text_inputs[0][0]

        _UpperCamelCase   :						Optional[int]    =							self.model(**__a       , **__a						)

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


  def  __SCREAMING_SNAKE_CASE					(		self :							Union[str, Any]       , __a :							Optional[Any]						)		->					Optional[Any]:
        _UpperCamelCase   :						Any    =							model_outputs.pop("candidate_labels"						)
        _UpperCamelCase   :						str    =							model_outputs["logits"][0]

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

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







"""simple docstring"""


from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding



class   __SCREAMING_SNAKE_CASE     (    _UpperCamelCase ):







  '''simple docstring'''


  SCREAMING_SNAKE_CASE__				:Optional[Any]						  =       ["image_processor", "tokenizer"]
  SCREAMING_SNAKE_CASE__				:Dict						  =       "AutoImageProcessor"
  SCREAMING_SNAKE_CASE__				:Optional[int]						  =       "AutoTokenizer"

  def __init__(		self :							Dict       , __a :							Tuple       , __a :							str						)		->					Optional[Any]:
        super().__init__(__a       , __a						)
        _UpperCamelCase   :						Tuple    =							self.image_processor

  def __call__(		self :							Tuple       , __a :							Tuple=None       , __a :							str=None       , __a :							Any=None       , **__a :							Optional[Any]						)		->					Union[str, Any]:

        if text is None and images is None:
              raise ValueError("You have to specify either text or images. Both cannot be none."						)

        if text is not None:
              _UpperCamelCase   :						List[str]    =							self.tokenizer(__a       , return_tensors=__a       , **__a						)

        if images is not None:
              _UpperCamelCase   :						str    =							self.image_processor(__a       , return_tensors=__a       , **__a						)

        if text is not None and images is not None:
              _UpperCamelCase   :						Optional[int]    =							image_features.pixel_values
              return encoding
        elif text is not None:
              return encoding
        else:
              return BatchEncoding(data=dict(**__a						)       , tensor_type=__a						)

  def  __SCREAMING_SNAKE_CASE					(		self :							Optional[int]       , *__a :							Optional[int]       , **__a :							Optional[int]						)		->					List[Any]:
        return self.tokenizer.batch_decode(*__a       , **__a						)

  def  __SCREAMING_SNAKE_CASE					(		self :							int       , *__a :							Any       , **__a :							List[Any]						)		->					Tuple:
        return self.tokenizer.decode(*__a       , **__a						)


  @property
  def  __SCREAMING_SNAKE_CASE					(		self :							str						)		->					Any:
        return ["input_ids", "attention_mask", "pixel_values"]
