


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




def       a__						(		snake_case__				:       bytes	,      snake_case__				:       int    ):
  _UpperCAmelCase		:       List[Any]       =    f'''{sampling_rate}'''
  _UpperCAmelCase		:       List[str]       =    """1"""
  _UpperCAmelCase		:       Union[str, Any]       =    """f32le"""
  _UpperCAmelCase		:       Dict       =    [
      """ffmpeg""",
      """-i""",
      """pipe:0""",
      """-ac""",
      ac,
      """-ar""",
      ar,
      """-f""",
      format_for_conversion,
      """-hide_banner""",
      """-loglevel""",
      """quiet""",
      """pipe:1""",
  ]

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

  _UpperCAmelCase		:       List[Any]       =    np.frombuffer(snake_case__	,      np.floataa    )
  if audio.shape[0] == 0:
    raise ValueError("""Malformed soundfile"""    )
  return audio





@add_end_docstrings(A    )
class 		_SCREAMING_SNAKE_CASE    (	A    ):


     def __init__(     self       ,				*A_       ,				**A_						):
       # Default, might be overriden by the model.config.
       _UpperCAmelCase		:       Optional[int]       =    5
       super().__init__(*A_       ,				**A_						)

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

       self.check_model_type(A_						)


     def __call__(     self       ,				A_       ,				**A_       ,				):
       return super().__call__(A_       ,				**A_						)


     def 					__snake_case(     self       ,				A_=None       ,				**A_						):
       # No parameters on this pipeline right now
       _UpperCAmelCase		:       int       =    {}
       if top_k is not None:
         if top_k > self.model.config.num_labels:
           _UpperCAmelCase		:       Tuple       =    self.model.config.num_labels
         _UpperCAmelCase		:       Dict       =    top_k
       return {}, {}, postprocess_params


     def 					__snake_case(     self       ,				A_						):
       if isinstance(A_       ,				A_						):
         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		:       str       =    requests.get(A_						).content
         else:
           with open(A_       ,				"""rb"""						) as f:
             _UpperCAmelCase		:       Dict       =    f.read()

       if isinstance(A_       ,				A_						):
         _UpperCAmelCase		:       List[Any]       =    ffmpeg_read(A_       ,				self.feature_extractor.sampling_rate						)

       if isinstance(A_       ,				A_						):
         # 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		:       Tuple       =    inputs.pop("""raw"""       ,				A_						)
         if _inputs is None:
           # Remove path which will not be used from `datasets`.
           inputs.pop("""path"""       ,				A_						)
           _UpperCAmelCase		:       Dict       =    inputs.pop("""array"""       ,				A_						)
         _UpperCAmelCase		:       Optional[int]       =    inputs.pop("""sampling_rate"""						)
         _UpperCAmelCase		:       Optional[Any]       =    _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		:       List[str]       =    F.resample(
               torch.from_numpy(A_						)       ,				A_       ,				self.feature_extractor.sampling_rate						).numpy()

       if not isinstance(A_       ,				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		:       Any       =    self.feature_extractor(
           A_       ,				sampling_rate=self.feature_extractor.sampling_rate       ,				return_tensors="""pt"""						)
       return processed


     def 					__snake_case(     self       ,				A_						):
       _UpperCAmelCase		:       Dict       =    self.model(**A_						)
       return model_outputs




     def 					__snake_case(     self       ,				A_       ,				A_=5						):
       _UpperCAmelCase		:       List[str]       =    model_outputs.logits[0].softmax(-1						)
       _UpperCAmelCase,_UpperCAmelCase		:       Dict       =    probs.topk(A_						)

       _UpperCAmelCase		:       Union[str, Any]       =    scores.tolist()
       _UpperCAmelCase		:       Dict       =    ids.tolist()

       _UpperCAmelCase		:       List[Any]       =    [{"""score""": score, """label""": self.model.config.idalabel[_id]} for score, _id in zip(A_       ,				A_						)]

       return labels









import os
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils import AddedToken
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
 from .tokenization_xlm_roberta import XLMRobertaTokenizer
else:
 SCREAMING_SNAKE_CASE__				:    Optional[Any]         =    None


SCREAMING_SNAKE_CASE__				:    Dict         =    logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__				:    List[Any]         =    {'vocab_file': 'sentencepiece.bpe.model', 'tokenizer_file': 'tokenizer.json'}

SCREAMING_SNAKE_CASE__				:    Any         =    {
    'vocab_file': {
        'xlm-roberta-base': 'https://huggingface.co/xlm-roberta-base/resolve/main/sentencepiece.bpe.model',
        'xlm-roberta-large': 'https://huggingface.co/xlm-roberta-large/resolve/main/sentencepiece.bpe.model',
        'xlm-roberta-large-finetuned-conll02-dutch': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll02-dutch/resolve/main/sentencepiece.bpe.model'
        ),
        'xlm-roberta-large-finetuned-conll02-spanish': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll02-spanish/resolve/main/sentencepiece.bpe.model'
        ),
        'xlm-roberta-large-finetuned-conll03-english': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll03-english/resolve/main/sentencepiece.bpe.model'
        ),
        'xlm-roberta-large-finetuned-conll03-german': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll03-german/resolve/main/sentencepiece.bpe.model'
        ),
    },
    'tokenizer_file': {
        'xlm-roberta-base': 'https://huggingface.co/xlm-roberta-base/resolve/main/tokenizer.json',
        'xlm-roberta-large': 'https://huggingface.co/xlm-roberta-large/resolve/main/tokenizer.json',
        'xlm-roberta-large-finetuned-conll02-dutch': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll02-dutch/resolve/main/tokenizer.json'
        ),
        'xlm-roberta-large-finetuned-conll02-spanish': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll02-spanish/resolve/main/tokenizer.json'
        ),
        'xlm-roberta-large-finetuned-conll03-english': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll03-english/resolve/main/tokenizer.json'
        ),
        'xlm-roberta-large-finetuned-conll03-german': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll03-german/resolve/main/tokenizer.json'
        ),
    },
}

SCREAMING_SNAKE_CASE__				:    Optional[int]         =    {
    'xlm-roberta-base': 512,
    'xlm-roberta-large': 512,
    'xlm-roberta-large-finetuned-conll02-dutch': 512,
    'xlm-roberta-large-finetuned-conll02-spanish': 512,
    'xlm-roberta-large-finetuned-conll03-english': 512,
    'xlm-roberta-large-finetuned-conll03-german': 512,
}


class 		_SCREAMING_SNAKE_CASE    (	A    ):
     __SCREAMING_SNAKE_CASE    				=						VOCAB_FILES_NAMES
     __SCREAMING_SNAKE_CASE    				=						PRETRAINED_VOCAB_FILES_MAP
     __SCREAMING_SNAKE_CASE    				=						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     __SCREAMING_SNAKE_CASE    				=						['''input_ids''', '''attention_mask''']
     __SCREAMING_SNAKE_CASE    				=						XLMRobertaTokenizer


     def __init__(     self       ,				A_=None       ,				A_=None       ,				A_="<s>"       ,				A_="</s>"       ,				A_="</s>"       ,				A_="<s>"       ,				A_="<unk>"       ,				A_="<pad>"       ,				A_="<mask>"       ,				**A_       ,				):
       # Mask token behave like a normal word, i.e. include the space before it
       _UpperCAmelCase		:       int       =    AddedToken(A_       ,				lstrip=A_       ,				rstrip=A_						) if isinstance(A_       ,				A_						) else mask_token

       super().__init__(
           A_       ,				tokenizer_file=A_       ,				bos_token=A_       ,				eos_token=A_       ,				sep_token=A_       ,				cls_token=A_       ,				unk_token=A_       ,				pad_token=A_       ,				mask_token=A_       ,				**A_       ,				)

       _UpperCAmelCase		:       Union[str, Any]       =    vocab_file
       _UpperCAmelCase		:       List[str]       =    False if not self.vocab_file else True


     def 					__snake_case(     self       ,				A_       ,				A_ = None						):

       if token_ids_a is None:
         return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
       _UpperCAmelCase		:       Optional[Any]       =    [self.cls_token_id]
       _UpperCAmelCase		:       Union[str, Any]       =    [self.sep_token_id]
       return cls + token_ids_a + sep + sep + token_ids_a + sep


     def 					__snake_case(     self       ,				A_       ,				A_ = None						):
       _UpperCAmelCase		:       Tuple       =    [self.sep_token_id]
       _UpperCAmelCase		:       Optional[int]       =    [self.cls_token_id]

       if token_ids_a is None:
         return len(cls + token_ids_a + sep						) * [0]
       return len(cls + token_ids_a + sep + sep + token_ids_a + sep						) * [0]




     def 					__snake_case(     self       ,				A_       ,				A_ = None						):
       if not self.can_save_slow_tokenizer:
         raise ValueError(
             """Your fast tokenizer does not have the necessary information to save the vocabulary for a slow """
             """tokenizer."""						)

       if not os.path.isdir(A_						):
         logger.error(F'''Vocabulary path ({save_directory}) should be a directory.'''						)
         return
       _UpperCAmelCase		:       List[str]       =    os.path.join(
           A_       ,				(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]						)

       if os.path.abspath(self.vocab_file						) != os.path.abspath(A_						):
         copyfile(self.vocab_file       ,				A_						)

       return (out_vocab_file,)





