





import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
  from transformers.pipelines.conversational import Conversation

SCREAMING_SNAKE_CASE__							:      str         =    logging.get_logger(__name__)


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

SCREAMING_SNAKE_CASE__							:      Tuple         =    {
    """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"""
    },
}

SCREAMING_SNAKE_CASE__							:      int         =    {"""facebook/blenderbot-3B""": 1_2_8}






@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def    _A	(   ):
   a__						:    Any   		=	(
       list(range(ord("!" )				,			ord("~" ) + 1 ) ) + list(range(ord("¡" )				,			ord("¬" ) + 1 ) ) + list(range(ord("®" )				,			ord("ÿ" ) + 1 ) )
   )
   a__						:    int   		=	bs[:]
   a__						:    int   		=	0
   for b in range(2**8 ):
      if b not in bs:
         bs.append(lowerCamelCase )
         cs.append(2**8 + n )
         n += 1
   a__						:    Optional[Any]   		=	[chr(lowerCamelCase ) for n in cs]
   return dict(zip(lowerCamelCase				,			lowerCamelCase ) )






def    _A	(   lowerCamelCase ):
   a__						:    Tuple   		=	set()
   a__						:    Tuple   		=	word[0]
   for char in word[1:]:
      pairs.add((prev_char, char) )
      a__						:    int   		=	char
   return pairs







class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    Tuple  		=					VOCAB_FILES_NAMES
   _UpperCamelCase			:    str  		=					PRETRAINED_VOCAB_FILES_MAP
   _UpperCamelCase			:    str  		=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   _UpperCamelCase			:    Optional[Any]  		=					["""input_ids""", """attention_mask"""]





   def __init__(			self       ,      snake_case       ,      snake_case       ,      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       ,      )					->	List[Any]:
      """simple docstring"""





      a__						:    List[str]   		=	AddedToken(snake_case       ,      lstrip=snake_case       ,      rstrip=snake_case   ) if isinstance(snake_case       ,      snake_case   ) else bos_token
      a__						:    Optional[int]   		=	AddedToken(snake_case       ,      lstrip=snake_case       ,      rstrip=snake_case   ) if isinstance(snake_case       ,      snake_case   ) else eos_token
      a__						:    Optional[Any]   		=	AddedToken(snake_case       ,      lstrip=snake_case       ,      rstrip=snake_case   ) if isinstance(snake_case       ,      snake_case   ) else sep_token
      a__						:    Tuple   		=	AddedToken(snake_case       ,      lstrip=snake_case       ,      rstrip=snake_case   ) if isinstance(snake_case       ,      snake_case   ) else cls_token
      a__						:    Optional[Any]   		=	AddedToken(snake_case       ,      lstrip=snake_case       ,      rstrip=snake_case   ) if isinstance(snake_case       ,      snake_case   ) else unk_token
      a__						:    Tuple   		=	AddedToken(snake_case       ,      lstrip=snake_case       ,      rstrip=snake_case   ) if isinstance(snake_case       ,      snake_case   ) else pad_token

      # Mask token behave like a normal word, i.e. include the space before it
      a__						:    Any   		=	AddedToken(snake_case       ,      lstrip=snake_case       ,      rstrip=snake_case   ) if isinstance(snake_case       ,      snake_case   ) else mask_token

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

      with open(snake_case       ,      encoding="utf-8"   ) as vocab_handle:
         a__						:    Optional[Any]   		=	json.load(snake_case   )
      a__						:    List[Any]   		=	{v: k for k, v in self.encoder.items()}
      a__						:    str   		=	errors  # how to handle errors in decoding
      a__						:    int   		=	bytes_to_unicode()
      a__						:    Optional[Any]   		=	{v: k for k, v in self.byte_encoder.items()}
      with open(snake_case       ,      encoding="utf-8"   ) as merges_handle:
         a__						:    Dict   		=	merges_handle.read().split("\n"   )[1:-1]
      a__						:    Optional[int]   		=	[tuple(merge.split()   ) for merge in bpe_merges]
      a__						:    Union[str, Any]   		=	dict(zip(snake_case       ,      range(len(snake_case   )   )   )   )
      a__						:    Dict   		=	{}
      a__						:    Tuple   		=	add_prefix_space

      # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
      a__						:    int   		=	re.compile(r"'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"   )





   @property
   # Copied from transformers.models.roberta.tokenization_roberta.RobertaTokenizer.vocab_size with Roberta->Blenderbot, RoBERTa->Blenderbot
   def 						_snake_case			(			self   )					->	Any:
      """simple docstring"""





      return len(self.encoder   )





   def 						_snake_case			(			self   )					->	Optional[int]:
      """simple docstring"""





      return dict(self.encoder       ,      **self.added_tokens_encoder   )





   def 						_snake_case			(			self       ,      snake_case   )					->	Any:
      """simple docstring"""





      if token in self.cache:
         return self.cache[token]
      a__						:    Union[str, Any]   		=	tuple(snake_case   )
      a__						:    Union[str, Any]   		=	get_pairs(snake_case   )

      if not pairs:
         return token

      while True:
         a__						:    int   		=	min(snake_case       ,      key=lambda snake_case   : self.bpe_ranks.get(snake_case       ,      float("inf"   )   )   )
         if bigram not in self.bpe_ranks:
            break
         a__		,  a__						:    int   		=	bigram
         a__						:    Dict   		=	[]
         a__						:    Optional[Any]   		=	0
         while i < len(snake_case   ):
            try:
               a__						:    Optional[Any]   		=	word.index(snake_case       ,      snake_case   )
            except ValueError:
               new_word.extend(word[i:]   )
               break
            else:
               new_word.extend(word[i:j]   )
               a__						:    List[Any]   		=	j

            if word[i] == first and i < len(snake_case   ) - 1 and word[i + 1] == second:
               new_word.append(first + second   )
               i += 2
            else:
               new_word.append(word[i]   )
               i += 1
         a__						:    Tuple   		=	tuple(snake_case   )
         a__						:    int   		=	new_word
         if len(snake_case   ) == 1:
            break
         else:
            a__						:    Union[str, Any]   		=	get_pairs(snake_case   )
      a__						:    List[Any]   		=	" ".join(snake_case   )
      a__						:    List[Any]   		=	word
      return word





   def 						_snake_case			(			self       ,      snake_case   )					->	Tuple:
      """simple docstring"""





      a__						:    List[Any]   		=	[]
      for token in re.findall(self.pat       ,      snake_case   ):
         a__						:    Optional[int]   		=	"".join(
             self.byte_encoder[b] for b in token.encode("utf-8"   )   )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
         bpe_tokens.extend(bpe_token for bpe_token in self.bpe(snake_case   ).split(" "   )   )
      return bpe_tokens





   def 						_snake_case			(			self       ,      snake_case   )					->	str:
      """simple docstring"""





      return self.encoder.get(snake_case       ,      self.encoder.get(self.unk_token   )   )





   def 						_snake_case			(			self       ,      snake_case   )					->	Optional[Any]:
      """simple docstring"""





      return self.decoder.get(snake_case   )





   def 						_snake_case			(			self       ,      snake_case   )					->	Optional[Any]:
      """simple docstring"""





      a__						:    Tuple   		=	"".join(snake_case   )
      a__						:    str   		=	bytearray([self.byte_decoder[c] for c in text]   ).decode("utf-8"       ,      errors=self.errors   )
      return text





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case = None   )					->	Tuple[str]:
      """simple docstring"""





      if not os.path.isdir(snake_case   ):
         logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""   )
         return
      a__						:    str   		=	os.path.join(
          snake_case       ,      (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]   )
      a__						:    Optional[Any]   		=	os.path.join(
          snake_case       ,      (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]   )

      with open(snake_case       ,      "w"       ,      encoding="utf-8"   ) as f:
         f.write(json.dumps(self.encoder       ,      indent=2       ,      sort_keys=snake_case       ,      ensure_ascii=snake_case   ) + "\n"   )

      a__						:    Union[str, Any]   		=	0
      with open(snake_case       ,      "w"       ,      encoding="utf-8"   ) as writer:
         writer.write("#version: 0.2\n"   )
         for bpe_tokens, token_index in sorted(self.bpe_ranks.items()       ,      key=lambda snake_case   : kv[1]   ):
            if index != token_index:
               logger.warning(
                   F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
                   " Please check that the tokenizer is not corrupted!"   )
               a__						:    Optional[int]   		=	token_index
            writer.write(" ".join(snake_case   ) + "\n"   )
            index += 1

      return vocab_file, merge_file





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case = None       ,      snake_case = False   )					->	List[int]:
      """simple docstring"""





      if already_has_special_tokens:
         return super().get_special_tokens_mask(
             token_ids_a=snake_case       ,      token_ids_a=snake_case       ,      already_has_special_tokens=snake_case   )

      if token_ids_a is None:
         return [1] + ([0] * len(snake_case   )) + [1]
      return [1] + ([0] * len(snake_case   )) + [1, 1] + ([0] * len(snake_case   )) + [1]





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case = None   )					->	List[int]:
      """simple docstring"""





      a__						:    Any   		=	[self.sep_token_id]
      a__						:    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 						_snake_case			(			self       ,      snake_case       ,      snake_case=False       ,      **snake_case   )					->	str:
      """simple docstring"""





      a__						:    int   		=	kwargs.pop("add_prefix_space"       ,      self.add_prefix_space   )
      if (is_split_into_words or add_prefix_space) and (len(snake_case   ) > 0 and not text[0].isspace()):
         a__						:    Dict   		=	" " + text
      return (text, kwargs)





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case = None   )					->	Dict:
      """simple docstring"""





      return token_ids_a + [self.eos_token_id]





   def 						_snake_case			(			self       ,      snake_case   )					->	List[int]:
      """simple docstring"""





      a__						:    List[Any]   		=	[]
      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   )

      a__						:    List[Any]   		=	"  ".join(snake_case   )
      a__						:    List[str]   		=	self.encode(snake_case   )
      if len(snake_case   ) > self.model_max_length:
         a__						:    str   		=	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








import warnings
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


SCREAMING_SNAKE_CASE__							:      Optional[int]         =    logging.get_logger(__name__)







class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    Tuple  		=					["""input_values""", """attention_mask"""]





   def __init__(			self       ,      snake_case = 1       ,      snake_case = 16_000       ,      snake_case = 0.0       ,      snake_case = False       ,      snake_case = 80       ,      snake_case = 16       ,      snake_case = 64       ,      snake_case = "hann_window"       ,      snake_case = 1.0       ,      snake_case = 80       ,      snake_case = 7_600       ,      snake_case = 1E-10       ,      snake_case = 2       ,      snake_case = True       ,      **snake_case       ,      )					->	Dict:
      """simple docstring"""





      super().__init__(feature_size=snake_case       ,      sampling_rate=snake_case       ,      padding_value=snake_case       ,      **snake_case   )
      a__						:    Any   		=	do_normalize
      a__						:    List[str]   		=	return_attention_mask

      a__						:    List[Any]   		=	num_mel_bins
      a__						:    List[str]   		=	hop_length
      a__						:    int   		=	win_length
      a__						:    List[Any]   		=	win_function
      a__						:    List[str]   		=	frame_signal_scale
      a__						:    List[Any]   		=	fmin
      a__						:    Optional[Any]   		=	fmax
      a__						:    Union[str, Any]   		=	mel_floor
      a__						:    Union[str, Any]   		=	reduction_factor

      a__						:    List[str]   		=	win_length * sampling_rate // 1_000
      a__						:    List[Any]   		=	hop_length * sampling_rate // 1_000
      a__						:    List[Any]   		=	optimal_fft_length(self.sample_size   )
      a__						:    Dict   		=	(self.n_fft // 2) + 1

      a__						:    str   		=	window_function(window_length=self.sample_size       ,      name=self.win_function       ,      periodic=snake_case   )

      a__						:    Tuple   		=	mel_filter_bank(
          num_frequency_bins=self.n_freqs       ,      num_mel_filters=self.num_mel_bins       ,      min_frequency=self.fmin       ,      max_frequency=self.fmax       ,      sampling_rate=self.sampling_rate       ,      norm="slaney"       ,      mel_scale="slaney"       ,      )

      if frame_signal_scale != 1.0:
         warnings.warn(
             "The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"       ,      snake_case       ,      )
      if reduction_factor != 2.0:
         warnings.warn(
             "The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"       ,      snake_case       ,      )





   @staticmethod
   # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
   def 						_snake_case			(			snake_case       ,      snake_case       ,      snake_case = 0.0   )					->	List[np.ndarray]:
      """simple docstring"""





      if attention_mask is not None:
         a__						:    Tuple   		=	np.array(snake_case       ,      np.intaa   )
         a__						:    List[str]   		=	[]

         for vector, length in zip(snake_case       ,      attention_mask.sum(-1   )   ):
            a__						:    List[Any]   		=	(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7   )
            if length < normed_slice.shape[0]:
               a__						:    List[str]   		=	padding_value

            normed_input_values.append(snake_case   )
      else:
         a__						:    Any   		=	[(x - x.mean()) / np.sqrt(x.var() + 1E-7   ) for x in input_values]

      return normed_input_values





   def 						_snake_case			(			self       ,      snake_case       ,      )					->	np.ndarray:
      """simple docstring"""





      a__						:    str   		=	spectrogram(
          snake_case       ,      window=self.window       ,      frame_length=self.sample_size       ,      hop_length=self.sample_stride       ,      fft_length=self.n_fft       ,      mel_filters=self.mel_filters       ,      mel_floor=self.mel_floor       ,      log_mel="log10"       ,      )
      return log_mel_spec.T





   def __call__(			self       ,      snake_case = None       ,      snake_case = None       ,      snake_case = False       ,      snake_case = None       ,      snake_case = False       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      **snake_case       ,      )					->	BatchFeature:
      """simple docstring"""





      if audio is None and audio_target is None:
         raise ValueError("You must provide either `audio` or `audio_target` values."   )

      if sampling_rate is not None:
         if sampling_rate != self.sampling_rate:
            raise ValueError(
                F"""The model corresponding to this feature extractor: {self} was trained using a sampling rate of"""
                F""" {self.sampling_rate}. Please make sure that the provided audio input was sampled with"""
                F""" {self.sampling_rate} and not {sampling_rate}."""   )
      else:
         logger.warning(
             "It is strongly recommended to pass the ``sampling_rate`` argument to this function. "
             "Failing to do so can result in silent errors that might be hard to debug."   )

      if audio is not None:
         a__						:    Dict   		=	self._process_audio(
             snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      **snake_case       ,      )
      else:
         a__						:    Optional[int]   		=	None

      if audio_target is not None:
         a__						:    List[Any]   		=	self._process_audio(
             snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      **snake_case       ,      )

         if inputs is None:
            return inputs_target
         else:
            a__						:    Tuple   		=	inputs_target["input_values"]
            a__						:    Tuple   		=	inputs_target.get("attention_mask"   )
            if decoder_attention_mask is not None:
               a__						:    Tuple   		=	decoder_attention_mask

      return inputs





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case = False       ,      snake_case = False       ,      snake_case = None       ,      snake_case = False       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      **snake_case       ,      )					->	BatchFeature:
      """simple docstring"""





      a__						:    Optional[int]   		=	isinstance(snake_case       ,      np.ndarray   ) and len(speech.shape   ) > 1
      if is_batched_numpy and len(speech.shape   ) > 2:
         raise ValueError(F"""Only mono-channel audio is supported for input to {self}"""   )
      a__						:    List[Any]   		=	is_batched_numpy or (
          isinstance(snake_case       ,      (list, tuple)   ) and (isinstance(speech[0]       ,      (np.ndarray, tuple, list)   ))
      )

      if is_batched:
         a__						:    int   		=	[np.asarray(snake_case       ,      dtype=np.floataa   ) for speech in speech]
      elif not is_batched and not isinstance(snake_case       ,      np.ndarray   ):
         a__						:    Any   		=	np.asarray(snake_case       ,      dtype=np.floataa   )
      elif isinstance(snake_case       ,      np.ndarray   ) and speech.dtype is np.dtype(np.floataa   ):
         a__						:    List[Any]   		=	speech.astype(np.floataa   )

      # always return batch
      if not is_batched:
         a__						:    Union[str, Any]   		=	[speech]

      # needed to make pad() work on spectrogram inputs
      a__						:    Optional[Any]   		=	self.feature_size

      # convert into correct format for padding
      if is_target:
         a__						:    List[str]   		=	[self._extract_mel_features(snake_case   ) for waveform in speech]
         a__						:    Optional[Any]   		=	BatchFeature({"input_values": features}   )
         a__						:    str   		=	self.num_mel_bins
      else:
         a__						:    int   		=	BatchFeature({"input_values": speech}   )

      a__						:    int   		=	self.pad(
          snake_case       ,      padding=snake_case       ,      max_length=snake_case       ,      truncation=snake_case       ,      pad_to_multiple_of=snake_case       ,      return_attention_mask=snake_case       ,      **snake_case       ,      )

      a__						:    Any   		=	feature_size_hack

      # convert input values to correct format
      a__						:    Tuple   		=	padded_inputs["input_values"]
      if not isinstance(input_values[0]       ,      np.ndarray   ):
         a__						:    int   		=	[np.asarray(snake_case       ,      dtype=np.floataa   ) for array in input_values]
      elif (
          not isinstance(snake_case       ,      np.ndarray   )
          and isinstance(input_values[0]       ,      np.ndarray   )
          and input_values[0].dtype is np.dtype(np.floataa   )
      ):
         a__						:    Union[str, Any]   		=	[array.astype(np.floataa   ) for array in input_values]
      elif isinstance(snake_case       ,      np.ndarray   ) and input_values.dtype is np.dtype(np.floataa   ):
         a__						:    Optional[int]   		=	input_values.astype(np.floataa   )

      # convert attention_mask to correct format
      a__						:    Optional[int]   		=	padded_inputs.get("attention_mask"   )
      if attention_mask is not None:
         a__						:    Tuple   		=	[np.asarray(snake_case       ,      dtype=np.intaa   ) for array in attention_mask]

      # zero-mean and unit-variance normalization
      if not is_target and self.do_normalize:
         a__						:    Any   		=	(
             attention_mask
             if self._get_padding_strategies(snake_case       ,      max_length=snake_case   ) is not PaddingStrategy.DO_NOT_PAD
             else None
         )
         a__						:    Optional[Any]   		=	self.zero_mean_unit_var_norm(
             padded_inputs["input_values"]       ,      attention_mask=snake_case       ,      padding_value=self.padding_value   )

      if return_tensors is not None:
         a__						:    int   		=	padded_inputs.convert_to_tensors(snake_case   )

      return padded_inputs





   def 						_snake_case			(			self   )					->	Dict[str, Any]:
      """simple docstring"""





      a__						:    int   		=	super().to_dict()

      # Don't serialize these as they are derived from the other properties.
      a__						:    str   		=	["window", "mel_filters", "sample_size", "sample_stride", "n_fft", "n_freqs"]
      for name in names:
         if name in output:
            del output[name]

      return output

