





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

A__       :     Union[str, Any]          =logging.get_logger(__name__)
def      UpperCamelCase__   (	lowerCAmelCase							,			lowerCAmelCase						):





      """simple docstring"""


      _lowerCAmelCase							  =							f"{sampling_rate}"
      _lowerCAmelCase							  =							"""1"""
      _lowerCAmelCase							  =							"""f32le"""
      _lowerCAmelCase							  =							[
          """ffmpeg""",
          """-i""",
          """pipe:0""",
          """-ac""",
          ac,
          """-ar""",
          ar,
          """-f""",
          format_for_conversion,
          """-hide_banner""",
          """-loglevel""",
          """quiet""",
          """pipe:1""",
      ]

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

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





@add_end_docstrings(snake_case_     )
class 				UpperCAmelCase  (    snake_case_     ):




      def __init__(    self    :				List[str]    ,    *__snake_case    :				str    ,    **__snake_case    :				List[Any]						)       ->   int:
            # Default, might be overriden by the model.config.
            _lowerCAmelCase							  =							5
            super().__init__(*__snake_case    ,    **__snake_case						)

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

            self.check_model_type(__snake_case						)




      def __call__(    self    :				str    ,    __snake_case    :				Union[np.ndarray, bytes, str]    ,    **__snake_case    :				Tuple    ,    )       ->   int:
            return super().__call__(__snake_case    ,    **__snake_case						)




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




      def        lowercase__					(    self    :				List[str]    ,    __snake_case    :				str						)       ->   Any:
            if isinstance(__snake_case    ,    __snake_case						):
                  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
                        _lowerCAmelCase							  =							requests.get(__snake_case						).content
                  else:
                        with open(__snake_case    ,    """rb"""						) as f:
                              _lowerCAmelCase							  =							f.read()

            if isinstance(__snake_case    ,    __snake_case						):
                  _lowerCAmelCase							  =							ffmpeg_read(__snake_case    ,    self.feature_extractor.sampling_rate						)

            if isinstance(__snake_case    ,    __snake_case						):
                  # 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"""						)

                  _lowerCAmelCase							  =							inputs.pop("""raw"""    ,    __snake_case						)
                  if _inputs is None:
                        # Remove path which will not be used from `datasets`.
                        inputs.pop("""path"""    ,    __snake_case						)
                        _lowerCAmelCase							  =							inputs.pop("""array"""    ,    __snake_case						)
                  _lowerCAmelCase							  =							inputs.pop("""sampling_rate"""						)
                  _lowerCAmelCase							  =							_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`."""						)

                        _lowerCAmelCase							  =							F.resample(
                            torch.from_numpy(__snake_case						)    ,    __snake_case    ,    self.feature_extractor.sampling_rate						).numpy()

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

            _lowerCAmelCase							  =							self.feature_extractor(
                __snake_case    ,    sampling_rate=self.feature_extractor.sampling_rate    ,    return_tensors="""pt"""						)
            return processed




      def        lowercase__					(    self    :				List[Any]    ,    __snake_case    :				Dict						)       ->   str:
            _lowerCAmelCase							  =							self.model(**__snake_case						)
            return model_outputs




      def        lowercase__					(    self    :				int    ,    __snake_case    :				Any    ,    __snake_case    :				Tuple=5						)       ->   List[Any]:
            _lowerCAmelCase							  =							model_outputs.logits[0].softmax(-1						)
            _lowerCAmelCase  ,							_lowerCAmelCase							  =							probs.topk(__snake_case						)

            _lowerCAmelCase							  =							scores.tolist()
            _lowerCAmelCase							  =							ids.tolist()

            _lowerCAmelCase							  =							[{"""score""": score, """label""": self.model.config.idalabel[_id]} for score, _id in zip(__snake_case    ,    __snake_case						)]

            return labels













'''simple docstring'''






import math

import torch
from torch import nn
from torch.nn import init

from .binarizer import MagnitudeBinarizer, ThresholdBinarizer, TopKBinarizer




class 				UpperCAmelCase  (    nn.Linear     ):




      def __init__(    self    :				str    ,    __snake_case    :				int    ,    __snake_case    :				int    ,    __snake_case    :				bool = True    ,    __snake_case    :				str = "constant"    ,    __snake_case    :				float = 0.0    ,    __snake_case    :				str = "topK"    ,    )       ->   int:
            super(__snake_case    ,    self						).__init__(in_features=__snake_case    ,    out_features=__snake_case    ,    bias=__snake_case						)
            assert pruning_method in ["topK", "threshold", "sigmoied_threshold", "magnitude", "l0"]
            _lowerCAmelCase							  =							pruning_method

            if self.pruning_method in ["topK", "threshold", "sigmoied_threshold", "l0"]:
                  _lowerCAmelCase							  =							mask_scale
                  _lowerCAmelCase							  =							mask_init
                  _lowerCAmelCase							  =							nn.Parameter(torch.empty(self.weight.size()						)						)
                  self.init_mask()




      def        lowercase__					(    self    :				Any						)       ->   str:
            if self.mask_init == "constant":
                  init.constant_(self.mask_scores    ,    val=self.mask_scale						)
            elif self.mask_init == "uniform":
                  init.uniform_(self.mask_scores    ,    a=-self.mask_scale    ,    b=self.mask_scale						)
            elif self.mask_init == "kaiming":
                  init.kaiming_uniform_(self.mask_scores    ,    a=math.sqrt(5						)						)




      def        lowercase__					(    self    :				Dict    ,    __snake_case    :				torch.tensor    ,    __snake_case    :				float						)       ->   Union[str, Any]:
            # Get the mask
            if self.pruning_method == "topK":
                  _lowerCAmelCase							  =							TopKBinarizer.apply(self.mask_scores    ,    __snake_case						)
            elif self.pruning_method in ["threshold", "sigmoied_threshold"]:
                  _lowerCAmelCase							  =							"""sigmoied""" in self.pruning_method
                  _lowerCAmelCase							  =							ThresholdBinarizer.apply(self.mask_scores    ,    __snake_case    ,    __snake_case						)
            elif self.pruning_method == "magnitude":
                  _lowerCAmelCase							  =							MagnitudeBinarizer.apply(self.weight    ,    __snake_case						)
            elif self.pruning_method == "l0":
                  _lowerCAmelCase  ,							_lowerCAmelCase  ,							_lowerCAmelCase							  =							-0.1, 1.1, 2 / 3
                  if self.training:
                        _lowerCAmelCase							  =							torch.zeros_like(self.mask_scores						).uniform_().clamp(0.00_01    ,    0.99_99						)
                        _lowerCAmelCase							  =							torch.sigmoid((u.log() - (1 - u).log() + self.mask_scores) / b						)
                  else:
                        _lowerCAmelCase							  =							torch.sigmoid(self.mask_scores						)
                  _lowerCAmelCase							  =							s * (r - l) + l
                  _lowerCAmelCase							  =							s_bar.clamp(min=0.0    ,    max=1.0						)
            # Mask weights with computed mask
            _lowerCAmelCase							  =							mask * self.weight
            # Compute output (linear layer) with masked weights
            return nn.functional.linear(__snake_case    ,    __snake_case    ,    self.bias						)






