

import json
from typing import List, Optional, Tuple

from tokenizers import pre_tokenizers, processors

from ...tokenization_utils_base import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roberta import RobertaTokenizer


lowercase_   							=  logging.get_logger(__name__)

lowercase_   							=  {'vocab_file': 'vocab.json', 'merges_file': 'merges.txt', 'tokenizer_file': 'tokenizer.json'}

lowercase_   							=  {
    'vocab_file': {
        'roberta-base': 'https://huggingface.co/roberta-base/resolve/main/vocab.json',
        'roberta-large': 'https://huggingface.co/roberta-large/resolve/main/vocab.json',
        'roberta-large-mnli': 'https://huggingface.co/roberta-large-mnli/resolve/main/vocab.json',
        'distilroberta-base': 'https://huggingface.co/distilroberta-base/resolve/main/vocab.json',
        'roberta-base-openai-detector': 'https://huggingface.co/roberta-base-openai-detector/resolve/main/vocab.json',
        'roberta-large-openai-detector': (
            'https://huggingface.co/roberta-large-openai-detector/resolve/main/vocab.json'
        ),
    },
    'merges_file': {
        'roberta-base': 'https://huggingface.co/roberta-base/resolve/main/merges.txt',
        'roberta-large': 'https://huggingface.co/roberta-large/resolve/main/merges.txt',
        'roberta-large-mnli': 'https://huggingface.co/roberta-large-mnli/resolve/main/merges.txt',
        'distilroberta-base': 'https://huggingface.co/distilroberta-base/resolve/main/merges.txt',
        'roberta-base-openai-detector': 'https://huggingface.co/roberta-base-openai-detector/resolve/main/merges.txt',
        'roberta-large-openai-detector': (
            'https://huggingface.co/roberta-large-openai-detector/resolve/main/merges.txt'
        ),
    },
    'tokenizer_file': {
        'roberta-base': 'https://huggingface.co/roberta-base/resolve/main/tokenizer.json',
        'roberta-large': 'https://huggingface.co/roberta-large/resolve/main/tokenizer.json',
        'roberta-large-mnli': 'https://huggingface.co/roberta-large-mnli/resolve/main/tokenizer.json',
        'distilroberta-base': 'https://huggingface.co/distilroberta-base/resolve/main/tokenizer.json',
        'roberta-base-openai-detector': (
            'https://huggingface.co/roberta-base-openai-detector/resolve/main/tokenizer.json'
        ),
        'roberta-large-openai-detector': (
            'https://huggingface.co/roberta-large-openai-detector/resolve/main/tokenizer.json'
        ),
    },
}

lowercase_   							=  {
    'roberta-base': 5_1_2,
    'roberta-large': 5_1_2,
    'roberta-large-mnli': 5_1_2,
    'distilroberta-base': 5_1_2,
    'roberta-base-openai-detector': 5_1_2,
    'roberta-large-openai-detector': 5_1_2,
}





class 					__lowerCAmelCase	(   SCREAMING_SNAKE_CASE				):
  _a		     =    VOCAB_FILES_NAMES
  _a		     =    PRETRAINED_VOCAB_FILES_MAP
  _a		     =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  _a		     =    ["""input_ids""", """attention_mask"""]
  _a		     =    RobertaTokenizer




  def __init__(							self				,  lowerCAmelCase=None				,  lowerCAmelCase=None				,  lowerCAmelCase=None				,  lowerCAmelCase="replace"				,  lowerCAmelCase="<s>"				,  lowerCAmelCase="</s>"				,  lowerCAmelCase="</s>"				,  lowerCAmelCase="<s>"				,  lowerCAmelCase="<unk>"				,  lowerCAmelCase="<pad>"				,  lowerCAmelCase="<mask>"				,  lowerCAmelCase=False				,  lowerCAmelCase=True				,  **lowerCAmelCase				,  )    ->					List[str]:
        '''simple docstring'''






        super().__init__(
            lowerCAmelCase				,  lowerCAmelCase				,  tokenizer_file=lowerCAmelCase				,  errors=lowerCAmelCase				,  bos_token=lowerCAmelCase				,  eos_token=lowerCAmelCase				,  sep_token=lowerCAmelCase				,  cls_token=lowerCAmelCase				,  unk_token=lowerCAmelCase				,  pad_token=lowerCAmelCase				,  mask_token=lowerCAmelCase				,  add_prefix_space=lowerCAmelCase				,  trim_offsets=lowerCAmelCase				,  **lowerCAmelCase				,  )

        _lowercase      			=json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()   )
        if pre_tok_state.get('add_prefix_space'				,  lowerCAmelCase   ) != add_prefix_space:
              _lowercase      			=getattr(lowerCAmelCase				,  pre_tok_state.pop('type'   )   )
              _lowercase      			=add_prefix_space
              _lowercase      			=pre_tok_class(**lowerCAmelCase   )

        _lowercase      			=add_prefix_space

        _lowercase      			='post_processor'
        _lowercase      			=getattr(self.backend_tokenizer				,  lowerCAmelCase				,  lowerCAmelCase   )
        if tokenizer_component_instance:
              _lowercase      			=json.loads(tokenizer_component_instance.__getstate__()   )

              # The lists 'sep' and 'cls' must be cased in tuples for the object `post_processor_class`
              if "sep" in state:
                    _lowercase      			=tuple(state['sep']   )
              if "cls" in state:
                    _lowercase      			=tuple(state['cls']   )

              _lowercase      			=False

              if state.get('add_prefix_space'				,  lowerCAmelCase   ) != add_prefix_space:
                    _lowercase      			=add_prefix_space
                    _lowercase      			=True

              if state.get('trim_offsets'				,  lowerCAmelCase   ) != trim_offsets:
                    _lowercase      			=trim_offsets
                    _lowercase      			=True

              if changes_to_apply:
                    _lowercase      			=getattr(lowerCAmelCase				,  state.pop('type'   )   )
                    _lowercase      			=component_class(**lowerCAmelCase   )
                    setattr(self.backend_tokenizer				,  lowerCAmelCase				,  lowerCAmelCase   )




  @property
  def     A__	(							self   )    ->					str:
        '''simple docstring'''






        if self._mask_token is None:
              if self.verbose:
                    logger.error('Using mask_token, but it is not set yet.'   )
              return None
        return str(self._mask_token   )




  @mask_token.setter
  def     A__	(							self				,  lowerCAmelCase   )    ->					Dict:
        '''simple docstring'''






        _lowercase      			=AddedToken(lowerCAmelCase				,  lstrip=lowerCAmelCase				,  rstrip=lowerCAmelCase   ) if isinstance(lowerCAmelCase				,  lowerCAmelCase   ) else value
        _lowercase      			=value




  def     A__	(							self				,  *lowerCAmelCase				,  **lowerCAmelCase   )    ->					BatchEncoding:
        '''simple docstring'''






        _lowercase      			=kwargs.get('is_split_into_words'				,  lowerCAmelCase   )
        assert self.add_prefix_space or not is_split_into_words, (
            F'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True '''
            "to use it with pretokenized inputs."
        )

        return super()._batch_encode_plus(*lowerCAmelCase				,  **lowerCAmelCase   )




  def     A__	(							self				,  *lowerCAmelCase				,  **lowerCAmelCase   )    ->					BatchEncoding:
        '''simple docstring'''






        _lowercase      			=kwargs.get('is_split_into_words'				,  lowerCAmelCase   )

        assert self.add_prefix_space or not is_split_into_words, (
            F'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True '''
            "to use it with pretokenized inputs."
        )

        return super()._encode_plus(*lowerCAmelCase				,  **lowerCAmelCase   )




  def     A__	(							self				,  lowerCAmelCase				,  lowerCAmelCase = None   )    ->					Tuple[str]:
        '''simple docstring'''






        _lowercase      			=self._tokenizer.model.save(lowerCAmelCase				,  name=lowerCAmelCase   )
        return tuple(lowerCAmelCase   )




  def     A__	(							self				,  lowerCAmelCase				,  lowerCAmelCase=None   )    ->					Optional[Any]:
        '''simple docstring'''






        _lowercase      			=[self.bos_token_id] + token_ids_a + [self.eos_token_id]
        if token_ids_a is None:
              return output

        return output + [self.eos_token_id] + token_ids_a + [self.eos_token_id]




  def     A__	(							self				,  lowerCAmelCase				,  lowerCAmelCase = None   )    ->					List[int]:
        '''simple docstring'''






        _lowercase      			=[self.sep_token_id]
        _lowercase      			=[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]





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


lowercase_   							=  logging.get_logger(__name__)





@add_end_docstrings(SCREAMING_SNAKE_CASE				)
class 					__lowerCAmelCase	(   SCREAMING_SNAKE_CASE				):




  def __init__(							self				,  **lowerCAmelCase   )    ->					Tuple:
        '''simple docstring'''






        super().__init__(**lowerCAmelCase   )

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




  def __call__(							self				,  lowerCAmelCase				,  **lowerCAmelCase   )    ->					int:
        '''simple docstring'''






        return super().__call__(lowerCAmelCase				,  **lowerCAmelCase   )




  def     A__	(							self				,  **lowerCAmelCase   )    ->					Optional[Any]:
        '''simple docstring'''






        _lowercase      			={}
        if "candidate_labels" in kwargs:
              _lowercase      			=kwargs['candidate_labels']
        if "hypothesis_template" in kwargs:
              _lowercase      			=kwargs['hypothesis_template']

        return preprocess_params, {}, {}




  def     A__	(							self				,  lowerCAmelCase				,  lowerCAmelCase=None				,  lowerCAmelCase="This is a sound of {}."   )    ->					Tuple:
        '''simple docstring'''






        if isinstance(lowerCAmelCase				,  lowerCAmelCase   ):
              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
                    _lowercase      			=requests.get(lowerCAmelCase   ).content
              else:
                    with open(lowerCAmelCase				,  'rb'   ) as f:
                          _lowercase      			=f.read()

        if isinstance(lowerCAmelCase				,  lowerCAmelCase   ):
              _lowercase      			=ffmpeg_read(lowerCAmelCase				,  self.feature_extractor.sampling_rate   )

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

        _lowercase      			=self.feature_extractor(
            [audio]				,  sampling_rate=self.feature_extractor.sampling_rate				,  return_tensors='pt'   )
        _lowercase      			=candidate_labels
        _lowercase      			=[hypothesis_template.format(lowerCAmelCase   ) for x in candidate_labels]
        _lowercase      			=self.tokenizer(lowerCAmelCase				,  return_tensors=self.framework				,  padding=lowerCAmelCase   )
        _lowercase      			=[text_inputs]
        return inputs




  def     A__	(							self				,  lowerCAmelCase   )    ->					Dict:
        '''simple docstring'''






        _lowercase      			=model_inputs.pop('candidate_labels'   )
        _lowercase      			=model_inputs.pop('text_inputs'   )
        if isinstance(text_inputs[0]				,  lowerCAmelCase   ):
              _lowercase      			=text_inputs[0]
        else:
              # Batching case.
              _lowercase      			=text_inputs[0][0]

        _lowercase      			=self.model(**lowerCAmelCase				,  **lowerCAmelCase   )

        _lowercase      			={
            'candidate_labels': candidate_labels,
            'logits': outputs.logits_per_audio,
        }
        return model_outputs




  def     A__	(							self				,  lowerCAmelCase   )    ->					Tuple:
        '''simple docstring'''






        _lowercase      			=model_outputs.pop('candidate_labels'   )
        _lowercase      			=model_outputs['logits'][0]

        if self.framework == "pt":
              _lowercase      			=logits.softmax(dim=0   )
              _lowercase      			=probs.tolist()
        else:
              raise ValueError('`tf` framework not supported.'   )

        _lowercase      			=[
            {'score': score, 'label': candidate_label}
            for score, candidate_label in sorted(zip(lowerCAmelCase				,  lowerCAmelCase   )				,  key=lambda lowerCAmelCase   : -x[0]   )
        ]
        return result


