


"""simple docstring"""







import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

__lowercase          = logging.get_logger(__name__)


def 	lowercase     (							A_    ,  A_			)-> np.array:
  '''simple docstring'''







  a		:						Optional[int] 				=     F'''{sampling_rate}'''
  a		:						Optional[int] 				=     "1"
  a		:						Optional[Any] 				=     "f32le"
  a		:						Union[str, Any] 				=     [
      "ffmpeg",
      "-i",
      "pipe:0",
      "-ac",
      ac,
      "-ar",
      ar,
      "-f",
      format_for_conversion,
      "-hide_banner",
      "-loglevel",
      "quiet",
      "pipe:1",
  ]

  try:
    a		:						Optional[int] 				=     subprocess.Popen(A_    ,  stdin=subprocess.PIPE    ,  stdout=subprocess.PIPE			)
  except FileNotFoundError:
    raise ValueError("ffmpeg was not found but is required to load audio files from filename"			)
  a		:						Optional[int] 				=     ffmpeg_process.communicate(A_			)
  a		:						List[str] 				=     output_stream[0]

  a		:						str 				=     np.frombuffer(A_    ,  np.floataa			)
  if audio.shape[0] == 0:
    raise ValueError("Malformed soundfile"			)
  return audio







@add_end_docstrings(_a    )
class 		_A    (    _a    ):







    """simple docstring"""




    def __init__(   self							:       Optional[Any]							,					*__UpperCAmelCase							:       str							,					**__UpperCAmelCase							:       List[Any]):
      # Default, might be overriden by the model.config.
      a		:						Optional[Any] 				=     5
      super().__init__(*__UpperCAmelCase							,					**__UpperCAmelCase)

      if self.framework != "pt":
        raise ValueError(f'''The {self.__class__} is only available in PyTorch.''')

      self.check_model_type(__UpperCAmelCase)






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






    def       __snake_case				(   self							:       List[Any]							,					__UpperCAmelCase							:       int=None							,					**__UpperCAmelCase							:       Any):
      # No parameters on this pipeline right now
      a		:						Optional[Any] 				=     {}
      if top_k is not None:
        if top_k > self.model.config.num_labels:
          a		:						List[Any] 				=     self.model.config.num_labels
        a		:						Union[str, Any] 				=     top_k
      return {}, {}, postprocess_params






    def       __snake_case				(   self							:       int							,					__UpperCAmelCase							:       str):
      if isinstance(__UpperCAmelCase							,					__UpperCAmelCase):
        if inputs.startswith("http://") or inputs.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		:						str 				=     requests.get(__UpperCAmelCase).content
        else:
          with open(__UpperCAmelCase							,					"rb") as f:
            a		:						Optional[int] 				=     f.read()

      if isinstance(__UpperCAmelCase							,					__UpperCAmelCase):
        a		:						Union[str, Any] 				=     ffmpeg_read(__UpperCAmelCase							,					self.feature_extractor.sampling_rate)

      if isinstance(__UpperCAmelCase							,					__UpperCAmelCase):
        # Accepting `"array"` which is the key defined in `datasets` for
        # better integration
        if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
          raise ValueError(
              "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
              "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
              "containing the sampling_rate associated with that array")

        a		:						Optional[int] 				=     inputs.pop("raw"							,					__UpperCAmelCase)
        if _inputs is None:
          # Remove path which will not be used from `datasets`.
          inputs.pop("path"							,					__UpperCAmelCase)
          a		:						List[Any] 				=     inputs.pop("array"							,					__UpperCAmelCase)
        a		:						Tuple 				=     inputs.pop("sampling_rate")
        a		:						Tuple 				=     _inputs
        if in_sampling_rate != self.feature_extractor.sampling_rate:
          import torch

          if is_torchaudio_available():
            from torchaudio import functional as F
          else:
            raise ImportError(
                "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
                "The torchaudio package can be installed through: `pip install torchaudio`.")

          a		:						List[Any] 				=     F.resample(
              torch.from_numpy(__UpperCAmelCase)							,					__UpperCAmelCase							,					self.feature_extractor.sampling_rate).numpy()

      if not isinstance(__UpperCAmelCase							,					np.ndarray):
        raise ValueError("We expect a numpy ndarray as input")
      if len(inputs.shape) != 1:
        raise ValueError("We expect a single channel audio input for AudioClassificationPipeline")

      a		:						Dict 				=     self.feature_extractor(
          __UpperCAmelCase							,					sampling_rate=self.feature_extractor.sampling_rate							,					return_tensors="pt")
      return processed






    def       __snake_case				(   self							:       int							,					__UpperCAmelCase							:       str):
      a		:						List[str] 				=     self.model(**__UpperCAmelCase)
      return model_outputs






    def       __snake_case				(   self							:       Dict							,					__UpperCAmelCase							:       Tuple							,					__UpperCAmelCase							:       Any=5):
      a		:						List[str] 				=     model_outputs.logits[0].softmax(-1)
      a				,						a		:						Union[str, Any] 				=     probs.topk(__UpperCAmelCase)

      a		:						Tuple 				=     scores.tolist()
      a		:						Dict 				=     ids.tolist()

      a		:						Optional[int] 				=     [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(__UpperCAmelCase							,					__UpperCAmelCase)]

      return labels










"""simple docstring"""







from collections import OrderedDict
from typing import Mapping

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


__lowercase          = logging.get_logger(__name__)

__lowercase          = {
    """squeezebert/squeezebert-uncased""": (
        """https://huggingface.co/squeezebert/squeezebert-uncased/resolve/main/config.json"""
    ),
    """squeezebert/squeezebert-mnli""": """https://huggingface.co/squeezebert/squeezebert-mnli/resolve/main/config.json""",
    """squeezebert/squeezebert-mnli-headless""": (
        """https://huggingface.co/squeezebert/squeezebert-mnli-headless/resolve/main/config.json"""
    ),
}






class 		_A    (    _a    ):







    """simple docstring"""




    UpperCAmelCase			:				Tuple      =			SQUEEZEBERT_PRETRAINED_CONFIG_ARCHIVE_MAP
    UpperCAmelCase			:				Any      =			"""squeezebert"""






    def __init__(   self							:       List[str]							,					__UpperCAmelCase							:       Tuple=30522							,					__UpperCAmelCase							:       Tuple=768							,					__UpperCAmelCase							:       int=12							,					__UpperCAmelCase							:       str=12							,					__UpperCAmelCase							:       List[Any]=3072							,					__UpperCAmelCase							:       Dict="gelu"							,					__UpperCAmelCase							:       List[Any]=0.1							,					__UpperCAmelCase							:       Optional[int]=0.1							,					__UpperCAmelCase							:       Optional[Any]=512							,					__UpperCAmelCase							:       int=2							,					__UpperCAmelCase							:       Dict=0.02							,					__UpperCAmelCase							:       Optional[int]=1e-12							,					__UpperCAmelCase							:       Optional[Any]=0							,					__UpperCAmelCase							:       Optional[Any]=768							,					__UpperCAmelCase							:       Tuple=4							,					__UpperCAmelCase							:       Dict=4							,					__UpperCAmelCase							:       str=4							,					__UpperCAmelCase							:       List[Any]=1							,					__UpperCAmelCase							:       Optional[int]=4							,					__UpperCAmelCase							:       Optional[int]=4							,					**__UpperCAmelCase							:       Tuple							,					):
      super().__init__(pad_token_id=__UpperCAmelCase							,					**__UpperCAmelCase)

      a		:						List[Any] 				=     vocab_size
      a		:						List[Any] 				=     hidden_size
      a		:						Dict 				=     num_hidden_layers
      a		:						int 				=     num_attention_heads
      a		:						Dict 				=     hidden_act
      a		:						Tuple 				=     intermediate_size
      a		:						int 				=     hidden_dropout_prob
      a		:						List[Any] 				=     attention_probs_dropout_prob
      a		:						Union[str, Any] 				=     max_position_embeddings
      a		:						Union[str, Any] 				=     type_vocab_size
      a		:						Optional[Any] 				=     initializer_range
      a		:						Any 				=     layer_norm_eps
      a		:						Any 				=     embedding_size
      a		:						Tuple 				=     q_groups
      a		:						List[str] 				=     k_groups
      a		:						str 				=     v_groups
      a		:						Optional[Any] 				=     post_attention_groups
      a		:						List[str] 				=     intermediate_groups
      a		:						Optional[int] 				=     output_groups






class 		_A    (    _a    ):







    """simple docstring"""




    @property
    def       __snake_case				(   self							:       Optional[Any]):
      if self.task == "multiple-choice":
        a		:						Optional[int] 				=     {0: "batch", 1: "choice", 2: "sequence"}
      else:
        a		:						Tuple 				=     {0: "batch", 1: "sequence"}
      return OrderedDict(
          [
              ("input_ids", dynamic_axis),
              ("attention_mask", dynamic_axis),
              ("token_type_ids", dynamic_axis),
          ])






