
import json
from typing import TYPE_CHECKING, 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_blenderbot import BlenderbotTokenizer


if TYPE_CHECKING:
  from transformers.pipelines.conversational import Conversation

lowercase  :					List[str]						    =							logging.get_logger(__name__)


lowercase  :					str						    =							{
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
    """tokenizer_config_file""": """tokenizer_config.json""",
}

lowercase  :					Optional[Any]						    =							{
    """vocab_file""": {"""facebook/blenderbot-3B""": """https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json"""},
    """merges_file""": {"""facebook/blenderbot-3B""": """https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt"""},
    """tokenizer_config_file""": {
        """facebook/blenderbot-3B""": """https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json"""
    },
}

lowercase  :					List[Any]						    =							{"""facebook/blenderbot-3B""": 128}

class __snake_case		(      lowerCAmelCase     ):
    _a		:						Tuple=				VOCAB_FILES_NAMES
    _a		:						Union[str, Any]=				PRETRAINED_VOCAB_FILES_MAP
    _a		:						Optional[int]=				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    _a		:						List[Any]=				["input_ids", "attention_mask"]
    _a		:						Tuple=				BlenderbotTokenizer
    def __init__(  self		,snake_case=None		,snake_case=None		,snake_case=None		,snake_case="replace"		,snake_case="<s>"		,snake_case="</s>"		,snake_case="</s>"		,snake_case="<s>"		,snake_case="<unk>"		,snake_case="<pad>"		,snake_case="<mask>"		,snake_case=False		,snake_case=True		,**snake_case		,):



           '''simple docstring'''







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

           lowercase							:     Tuple       	=      json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()      )
           if pre_tok_state.get("""add_prefix_space"""		,snake_case      ) != add_prefix_space:
                  lowercase							:     Union[str, Any]       	=      getattr(snake_case		,pre_tok_state.pop("""type"""      )      )
                  lowercase							:     str       	=      add_prefix_space
                  lowercase							:     Any       	=      pre_tok_class(**snake_case      )

           lowercase							:     Tuple       	=      add_prefix_space

           lowercase							:     int       	=      """post_processor"""
           lowercase							:     Dict       	=      getattr(self.backend_tokenizer		,snake_case		,snake_case      )
           if tokenizer_component_instance:
                  lowercase							:     int       	=      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							:     Any       	=      tuple(state["""sep"""]      )
                  if "cls" in state:
                         lowercase							:     Union[str, Any]       	=      tuple(state["""cls"""]      )

                  lowercase							:     int       	=      False

                  if state.get("""add_prefix_space"""		,snake_case      ) != add_prefix_space:
                         lowercase							:     List[str]       	=      add_prefix_space
                         lowercase							:     Union[str, Any]       	=      True

                  if state.get("""trim_offsets"""		,snake_case      ) != trim_offsets:
                         lowercase							:     str       	=      trim_offsets
                         lowercase							:     str       	=      True

                  if changes_to_apply:
                         lowercase							:     Union[str, Any]       	=      getattr(snake_case		,state.pop("""type"""      )      )
                         lowercase							:     int       	=      component_class(**snake_case      )
                         setattr(self.backend_tokenizer		,snake_case		,snake_case      )
    @property
    # Copied from transformers.models.roberta.tokenization_roberta_fast.RobertaTokenizerFast.mask_token with Roberta->Blenderbot, RoBERTa->Blenderbot
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''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       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           lowercase							:     str       	=      AddedToken(snake_case		,lstrip=snake_case		,rstrip=snake_case      ) if isinstance(snake_case		,snake_case      ) else value
           lowercase							:     List[Any]       	=      value
    def       _SCREAMING_SNAKE_CASE       (  self		,*snake_case		,**snake_case      ):



           '''simple docstring'''







           lowercase							:     Optional[Any]       	=      kwargs.get("""is_split_into_words"""		,snake_case      )
           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(*snake_case		,**snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self		,*snake_case		,**snake_case      ):



           '''simple docstring'''







           lowercase							:     List[str]       	=      kwargs.get("""is_split_into_words"""		,snake_case      )

           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(*snake_case		,**snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case = None      ):



           '''simple docstring'''







           lowercase							:     int       	=      self._tokenizer.model.save(snake_case		,name=snake_case      )
           return tuple(snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case = None      ):



           '''simple docstring'''







           lowercase							:     Optional[int]       	=      [self.sep_token_id]
           lowercase							:     List[str]       	=      [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       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case = None      ):



           '''simple docstring'''







           return token_ids_a + [self.eos_token_id]
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           lowercase							:     Dict       	=      []
           for is_user, text in conversation.iter_texts():
                  if is_user:
                         # We need to space prefix as it's being done within blenderbot
                         inputs.append(""" """ + text      )
                  else:
                         # Generated responses should contain them already.
                         inputs.append(snake_case      )

           lowercase							:     List[Any]       	=      """  """.join(snake_case      )
           lowercase							:     Optional[Any]       	=      self.encode(snake_case      )
           if len(snake_case      ) > self.model_max_length:
                  lowercase							:     Any       	=      input_ids[-self.model_max_length :]
                  logger.warning(f"Trimmed input from conversation as it was longer than {self.model_max_length} tokens."      )
           return input_ids


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

@add_end_docstrings(lowerCAmelCase     )
class __snake_case		(      lowerCAmelCase     ):
    def __init__(  self		,**snake_case      ):



           '''simple docstring'''







           super().__init__(**snake_case      )

           if self.framework != "pt":
                  raise ValueError(f"The {self.__class__} is only available in PyTorch."      )
           # No specific FOR_XXX available yet
    def __call__(  self		,snake_case		,**snake_case      ):



           '''simple docstring'''







           return super().__call__(snake_case		,**snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self		,**snake_case      ):



           '''simple docstring'''







           lowercase							:     Union[str, Any]       	=      {}
           if "candidate_labels" in kwargs:
                  lowercase							:     List[str]       	=      kwargs["""candidate_labels"""]
           if "hypothesis_template" in kwargs:
                  lowercase							:     Dict       	=      kwargs["""hypothesis_template"""]

           return preprocess_params, {}, {}
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case=None		,snake_case="This is a sound of {}."      ):



           '''simple docstring'''







           if isinstance(snake_case		,snake_case      ):
                  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							:     Optional[Any]       	=      requests.get(snake_case      ).content
                  else:
                         with open(snake_case		,"""rb"""      ) as f:
                                lowercase							:     Union[str, Any]       	=      f.read()

           if isinstance(snake_case		,snake_case      ):
                  lowercase							:     int       	=      ffmpeg_read(snake_case		,self.feature_extractor.sampling_rate      )

           if not isinstance(snake_case		,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							:     Dict       	=      self.feature_extractor(
               [audio]		,sampling_rate=self.feature_extractor.sampling_rate		,return_tensors="""pt"""      )
           lowercase							:     Tuple       	=      candidate_labels
           lowercase							:     Tuple       	=      [hypothesis_template.format(snake_case      ) for x in candidate_labels]
           lowercase							:     Optional[Any]       	=      self.tokenizer(snake_case		,return_tensors=self.framework		,padding=snake_case      )
           lowercase							:     Optional[Any]       	=      [text_inputs]
           return inputs
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           lowercase							:     List[str]       	=      model_inputs.pop("""candidate_labels"""      )
           lowercase							:     Dict       	=      model_inputs.pop("""text_inputs"""      )
           if isinstance(text_inputs[0]		,snake_case      ):
                  lowercase							:     List[Any]       	=      text_inputs[0]
           else:
                  # Batching case.
                  lowercase							:     Dict       	=      text_inputs[0][0]

           lowercase							:     Optional[Any]       	=      self.model(**snake_case		,**snake_case      )

           lowercase							:     Any       	=      {
               """candidate_labels""": candidate_labels,
               """logits""": outputs.logits_per_audio,
           }
           return model_outputs
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           lowercase							:     List[Any]       	=      model_outputs.pop("""candidate_labels"""      )
           lowercase							:     Any       	=      model_outputs["""logits"""][0]

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

           lowercase							:     Tuple       	=      [
               {"""score""": score, """label""": candidate_label}
               for score, candidate_label in sorted(zip(snake_case		,snake_case      )		,key=lambda snake_case      : -x[0]      )
           ]
           return result
