

"""simple docstring"""

from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


SCREAMING_SNAKE_CASE_         =  logging.get_logger(__name__)

SCREAMING_SNAKE_CASE_         =  {
    '''facebook/m2m100_418M''': '''https://huggingface.co/facebook/m2m100_418M/resolve/main/config.json''',
    # See all M2M100 models at https://huggingface.co/models?filter=m2m_100
}
class        _UpperCAmelCase							(						lowerCamelCase__							):
      __SCREAMING_SNAKE_CASE				:							List[Any]							 =		'm2m_100'
      __SCREAMING_SNAKE_CASE				:							int							 =		['past_key_values']
      __SCREAMING_SNAKE_CASE				:							List[str]							 =		{'num_attention_heads': 'encoder_attention_heads', 'hidden_size': 'd_model'}




      def __init__(							self				,    lowercase_=1_2_8_1_1_2				,    lowercase_=1_0_2_4				,    lowercase_=1_2				,    lowercase_=4_0_9_6				,    lowercase_=1_6				,    lowercase_=1_2				,    lowercase_=4_0_9_6				,    lowercase_=1_6				,    lowercase_=0.0_5				,    lowercase_=0.0_5				,    lowercase_=True				,    lowercase_=True				,    lowercase_="relu"				,    lowercase_=1_0_2_4				,    lowercase_=0.1				,    lowercase_=0.1				,    lowercase_=0.0				,    lowercase_=0.0_2				,    lowercase_=2				,    lowercase_=True				,    lowercase_=1				,    lowercase_=0				,    lowercase_=2				,    **lowercase_				,    )  ->   Dict:
          UpperCAmelCase       =   vocab_size
          UpperCAmelCase       =   max_position_embeddings
          UpperCAmelCase       =   d_model
          UpperCAmelCase       =   encoder_ffn_dim
          UpperCAmelCase       =   encoder_layers
          UpperCAmelCase       =   encoder_attention_heads
          UpperCAmelCase       =   decoder_ffn_dim
          UpperCAmelCase       =   decoder_layers
          UpperCAmelCase       =   decoder_attention_heads
          UpperCAmelCase       =   dropout
          UpperCAmelCase       =   attention_dropout
          UpperCAmelCase       =   activation_dropout
          UpperCAmelCase       =   activation_function
          UpperCAmelCase       =   init_std
          UpperCAmelCase       =   encoder_layerdrop
          UpperCAmelCase       =   decoder_layerdrop
          UpperCAmelCase       =   use_cache
          UpperCAmelCase       =   encoder_layers
          UpperCAmelCase       =   scale_embedding  # scale factor will be sqrt(d_model) if True

          super().__init__(
              pad_token_id=lowercase_				,    bos_token_id=lowercase_				,    eos_token_id=lowercase_				,    is_encoder_decoder=lowercase_				,    decoder_start_token_id=lowercase_				,    **lowercase_				,    )

class        _UpperCAmelCase							(						lowerCamelCase__							):




      @property
      def       a_ (							self							)  ->   Mapping[str, Mapping[int, str]]:
          UpperCAmelCase       =   OrderedDict(
              [
                  ('input_ids', {0: 'batch', 1: 'encoder_sequence'}),
                  ('attention_mask', {0: 'batch', 1: 'encoder_sequence'}),
              ]							)

          if self.use_past:
              UpperCAmelCase       =   {0: 'batch'}
              UpperCAmelCase       =   {0: 'batch', 1: 'past_decoder_sequence + sequence'}
          else:
              UpperCAmelCase       =   {0: 'batch', 1: 'decoder_sequence'}
              UpperCAmelCase       =   {0: 'batch', 1: 'decoder_sequence'}

          if self.use_past:
              self.fill_with_past_key_values_(lowercase_				,    direction='inputs'							)
          return common_inputs




      def       a_ (							self				,    lowercase_				,    lowercase_ = -1				,    lowercase_ = -1				,    lowercase_ = False				,    lowercase_ = None				,    )  ->   Mapping[str, Any]:
          UpperCAmelCase       =   compute_effective_axis_dimension(
              lowercase_				,    fixed_dimension=OnnxConfig.default_fixed_batch				,    num_token_to_add=0							)

          # If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
          UpperCAmelCase       =   tokenizer.num_special_tokens_to_add(lowercase_							)
          UpperCAmelCase       =   compute_effective_axis_dimension(
              lowercase_				,    fixed_dimension=OnnxConfig.default_fixed_sequence				,    num_token_to_add=lowercase_							)

          # Generate dummy inputs according to compute batch and sequence
          UpperCAmelCase       =   [' '.join([tokenizer.unk_token]							) * seq_length] * batch_size
          UpperCAmelCase       =   dict(tokenizer(lowercase_				,    return_tensors=lowercase_							)							)
          return common_inputs




      def       a_ (							self				,    lowercase_				,    lowercase_ = -1				,    lowercase_ = -1				,    lowercase_ = False				,    lowercase_ = None				,    )  ->   Mapping[str, Any]:
          UpperCAmelCase       =   self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
              lowercase_				,    lowercase_				,    lowercase_				,    lowercase_				,    lowercase_							)

          # Generate decoder inputs
          UpperCAmelCase       =   seq_length if not self.use_past else 1
          UpperCAmelCase       =   self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
              lowercase_				,    lowercase_				,    lowercase_				,    lowercase_				,    lowercase_							)
          UpperCAmelCase       =   {F"decoder_{name}": tensor for name, tensor in decoder_inputs.items()}
          UpperCAmelCase       =   dict(**lowercase_				,    **lowercase_							)

          if self.use_past:
              if not is_torch_available():
                  raise ValueError('Cannot generate dummy past_keys inputs without PyTorch installed.'							)
              else:
                  import torch
              UpperCAmelCase     ,							UpperCAmelCase       =   common_inputs['input_ids'].shape
              UpperCAmelCase       =   common_inputs['decoder_input_ids'].shape[1]
              UpperCAmelCase     ,							UpperCAmelCase       =   self.num_attention_heads
              UpperCAmelCase       =   (
                  batch,
                  num_encoder_attention_heads,
                  encoder_seq_length,
                  self._config.hidden_size // num_encoder_attention_heads,
              )
              UpperCAmelCase       =   decoder_seq_length + 3
              UpperCAmelCase       =   (
                  batch,
                  num_decoder_attention_heads,
                  decoder_past_length,
                  self._config.hidden_size // num_decoder_attention_heads,
              )

              UpperCAmelCase       =   torch.cat(
                  [common_inputs['decoder_attention_mask'], torch.ones(lowercase_				,    lowercase_							)]				,    dim=1							)

              UpperCAmelCase       =   []
              # If the number of encoder and decoder layers are present in the model configuration, both are considered
              UpperCAmelCase     ,							UpperCAmelCase       =   self.num_layers
              UpperCAmelCase       =   min(lowercase_				,    lowercase_							)
              UpperCAmelCase       =   max(lowercase_				,    lowercase_							) - min_num_layers
              UpperCAmelCase       =   'encoder' if num_encoder_layers > num_decoder_layers else 'decoder'

              for _ in range(lowercase_							):
                  common_inputs["past_key_values"].append(
                      (
                          torch.zeros(lowercase_							),
                          torch.zeros(lowercase_							),
                          torch.zeros(lowercase_							),
                          torch.zeros(lowercase_							),
                      )							)
              # TODO: test this.
              UpperCAmelCase       =   encoder_shape if remaining_side_name == 'encoder' else decoder_shape
              for _ in range(lowercase_				,    lowercase_							):
                  common_inputs["past_key_values"].append((torch.zeros(lowercase_							), torch.zeros(lowercase_							))							)
          return common_inputs


      __SCREAMING_SNAKE_CASE				:							List[Any]							 =		_generate_dummy_inputs_for_default_and_seqaseq_lm





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

SCREAMING_SNAKE_CASE_         =  logging.get_logger(__name__)




def       lowercase__  (   lowerCAmelCase			:   bytes	,			lowerCAmelCase			:   int	)	-> np.array:






    """simple docstring"""




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

    try:
        UpperCAmelCase       =   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'	)
    UpperCAmelCase       =   ffmpeg_process.communicate(lowerCAmelCase	)
    UpperCAmelCase       =   output_stream[0]

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







@add_end_docstrings(SCREAMING_SNAKE_CASE_							)
class        _UpperCAmelCase							(						SCREAMING_SNAKE_CASE_							):




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

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

          self.check_model_type(lowercase_							)




      def __call__(							self				,    lowercase_				,    **lowercase_				,    )  ->   Tuple:
          return super().__call__(lowercase_				,    **lowercase_							)




      def       a_ (							self				,    lowercase_=None				,    **lowercase_							)  ->   Optional[Any]:
          # No parameters on this pipeline right now
          UpperCAmelCase       =   {}
          if top_k is not None:
              if top_k > self.model.config.num_labels:
                  UpperCAmelCase       =   self.model.config.num_labels
              UpperCAmelCase       =   top_k
          return {}, {}, postprocess_params




      def       a_ (							self				,    lowercase_							)  ->   List[str]:
          if isinstance(lowercase_				,    lowercase_							):
              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
                  UpperCAmelCase       =   requests.get(lowercase_							).content
              else:
                  with open(lowercase_				,    'rb'							) as f:
                      UpperCAmelCase       =   f.read()

          if isinstance(lowercase_				,    lowercase_							):
              UpperCAmelCase       =   ffmpeg_read(lowercase_				,    self.feature_extractor.sampling_rate							)

          if isinstance(lowercase_				,    lowercase_							):
              # 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'							)

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

                  UpperCAmelCase       =   F.resample(
                      torch.from_numpy(lowercase_							)				,    lowercase_				,    self.feature_extractor.sampling_rate							).numpy()

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

          UpperCAmelCase       =   self.feature_extractor(
              lowercase_				,    sampling_rate=self.feature_extractor.sampling_rate				,    return_tensors='pt'							)
          return processed




      def       a_ (							self				,    lowercase_							)  ->   Any:
          UpperCAmelCase       =   self.model(**lowercase_							)
          return model_outputs




      def       a_ (							self				,    lowercase_				,    lowercase_=5							)  ->   int:
          UpperCAmelCase       =   model_outputs.logits[0].softmax(-1							)
          UpperCAmelCase     ,							UpperCAmelCase       =   probs.topk(lowercase_							)

          UpperCAmelCase       =   scores.tolist()
          UpperCAmelCase       =   ids.tolist()

          UpperCAmelCase       =   [{'score': score, 'label': self.model.config.idalabel[_id]} for score, _id in zip(lowercase_				,    lowercase_							)]

          return labels



