





"""simple docstring"""




from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


a						: List[Any]     	=			logging.get_logger(__name__)


class        __UpperCamelCase				(						_lowerCAmelCase						):
							lowerCamelCase   :					int      					=["input_features", "attention_mask"]






							def __init__(   self      ,					lowerCAmelCase__=80      ,					lowerCAmelCase__=1_6000      ,					lowerCAmelCase__=80      ,					lowerCAmelCase__=0.0      ,					lowerCAmelCase__=True      ,					lowerCAmelCase__=True      ,					lowerCAmelCase__=True      ,					**lowerCAmelCase__      ,					)	-> str:
								super().__init__(feature_size=_lowerCamelCase      ,					sampling_rate=_lowerCamelCase      ,					padding_value=_lowerCamelCase      ,					**_lowerCamelCase      )
								a      :		Optional[Any]	=				num_mel_bins
								a      :		Tuple	=				do_ceptral_normalize
								a      :		str	=				normalize_means
								a      :		Optional[int]	=				normalize_vars
								a      :		Dict	=				True






							def     __a   (   self      ,					lowerCAmelCase__      ,					)	-> List[str]:
								a      :		Optional[Any]	=				waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
								a      :		Optional[int]	=				torch.from_numpy(_lowerCamelCase      ).unsqueeze(0      )
								a      :		Dict	=				ta_kaldi.fbank(_lowerCamelCase      ,					num_mel_bins=self.num_mel_bins      ,					sample_frequency=self.sampling_rate      )
								return features.numpy()






							@staticmethod
							def     __a   (   lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__ = True      ,					lowerCAmelCase__ = True      ,					lowerCAmelCase__ = 0.0      ,					)	-> Optional[Any]:
								# make sure we normalize float32 arrays
								if normalize_means:
									a      :		int	=				x[:input_length].mean(axis=0      )
									a      :		int	=				np.subtract(_lowerCamelCase      ,					_lowerCamelCase      )
								if normalize_vars:
									a      :		Any	=				x[:input_length].std(axis=0      )
									a      :		Dict	=				np.divide(_lowerCamelCase      ,					_lowerCamelCase      )

								if input_length < x.shape[0]:
									a      :		List[Any]	=				padding_value

								# make sure array is in float32
								a      :		List[str]	=				x.astype(np.floataa      )

								return x






							def     __a   (   self      ,					lowerCAmelCase__      ,					lowerCAmelCase__ = None      )	-> int:
								a      :		str	=				attention_mask.sum(-1      ) if attention_mask is not None else [x.shape[0] for x in input_features]
								return [
								    self.utterance_cmvn(_lowerCamelCase      ,					_lowerCamelCase      ,					self.normalize_means      ,					self.normalize_vars      ,					self.padding_value      )
								    for x, n in zip(_lowerCamelCase      ,					_lowerCamelCase      )
								]






							def __call__(   self      ,					lowerCAmelCase__      ,					lowerCAmelCase__ = False      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = False      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					**lowerCAmelCase__      ,					)	-> List[Any]:

								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 `raw_speech` 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."      )

								a      :		Dict	=				isinstance(_lowerCamelCase      ,					np.ndarray      ) and len(raw_speech.shape      ) > 1
								if is_batched_numpy and len(raw_speech.shape      ) > 2:
									raise ValueError(f"""Only mono-channel audio is supported for input to {self}"""      )
								a      :		Any	=				is_batched_numpy or (
								    isinstance(_lowerCamelCase      ,					(list, tuple)      ) and (isinstance(raw_speech[0]      ,					(np.ndarray, tuple, list)      ))
								)

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

								# always return batch
								if not is_batched:
									a      :		Optional[Any]	=				[raw_speech]

								# extract fbank features
								a      :		Dict	=				[self._extract_fbank_features(_lowerCamelCase      ) for waveform in raw_speech]

								# convert into correct format for padding
								a      :		List[Any]	=				BatchFeature({"input_features": features}      )

								a      :		Union[str, Any]	=				self.pad(
								    _lowerCamelCase      ,					padding=_lowerCamelCase      ,					max_length=_lowerCamelCase      ,					truncation=_lowerCamelCase      ,					pad_to_multiple_of=_lowerCamelCase      ,					return_attention_mask=_lowerCamelCase      ,					**_lowerCamelCase      ,					)

								# make sure list is in array format
								a      :		List[Any]	=				padded_inputs.get("input_features"      )
								if isinstance(input_features[0]      ,					_lowerCamelCase      ):
									a      :		Tuple	=				[np.asarray(_lowerCamelCase      ,					dtype=np.floataa      ) for feature in input_features]

								a      :		Optional[Any]	=				padded_inputs.get("attention_mask"      )
								if attention_mask is not None:
									a      :		int	=				[np.asarray(_lowerCamelCase      ,					dtype=np.intaa      ) for array in attention_mask]

								# Utterance-level cepstral mean and variance normalization
								if self.do_ceptral_normalize:
									a      :		Union[str, Any]	=				(
									    np.array(_lowerCamelCase      ,					dtype=np.intaa      )
									    if self._get_padding_strategies(_lowerCamelCase      ,					max_length=_lowerCamelCase      ) is not PaddingStrategy.DO_NOT_PAD
									    else None
									)
									a      :		Optional[Any]	=				self.normalize(
									    padded_inputs["input_features"]      ,					attention_mask=_lowerCamelCase      )

								if return_tensors is not None:
									a      :		List[Any]	=				padded_inputs.convert_to_tensors(_lowerCamelCase      )

								return padded_inputs










"""simple docstring"""


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

lowercase_				     =     logging.get_logger(__name__)


lowercase_				     =     {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
    'tokenizer_config_file': 'tokenizer_config.json',
}

lowercase_				     =     {
    '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_				     =     {'facebook/blenderbot-3B': 128}



@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def 					lowerCAmelCase				(    ):




  """simple docstring"""







  __A									= (
      list(range(ord('''!''' )		,       ord('''~''' ) + 1 ) ) + list(range(ord('''¡''' )		,       ord('''¬''' ) + 1 ) ) + list(range(ord('''®''' )		,       ord('''ÿ''' ) + 1 ) )
  )
  __A									= bs[:]
  __A									= 0
  for b in range(2**8 ):
    if b not in bs:
      bs.append(__UpperCamelCase )
      cs.append(2**8 + n )
      n += 1
  __A									= [chr(__UpperCamelCase ) for n in cs]
  return dict(zip(__UpperCamelCase		,       __UpperCamelCase ) )



def 					lowerCAmelCase				(    __UpperCamelCase ):




  """simple docstring"""







  __A									= set()
  __A									= word[0]
  for char in word[1:]:
    pairs.add((prev_char, char) )
    __A									= char
  return pairs





class 					snake_case		(							_lowerCAmelCase						):


   '''simple docstring'''

   A_						:	Tuple						 = VOCAB_FILES_NAMES
   A_						:	Union[str, Any]						 = PRETRAINED_VOCAB_FILES_MAP
   A_						:	str						 = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   A_						:	Optional[Any]						 = ["input_ids", "attention_mask"]


   def __init__(	self :     Dict,       _lowerCamelCase :     Optional[Any],       _lowerCamelCase :     List[str],       _lowerCamelCase :     Dict="replace",       _lowerCamelCase :     Any="<s>",       _lowerCamelCase :     Optional[int]="</s>",       _lowerCamelCase :     Dict="</s>",       _lowerCamelCase :     List[Any]="<s>",       _lowerCamelCase :     List[str]="<unk>",       _lowerCamelCase :     str="<pad>",       _lowerCamelCase :     Any="<mask>",       _lowerCamelCase :     Any=False,       **_lowerCamelCase :     Tuple,       ):

     '''simple docstring'''
     __A									= AddedToken(_lowerCamelCase,       lstrip=_lowerCamelCase,       rstrip=_lowerCamelCase  ) if isinstance(_lowerCamelCase,       _lowerCamelCase  ) else bos_token
     __A									= AddedToken(_lowerCamelCase,       lstrip=_lowerCamelCase,       rstrip=_lowerCamelCase  ) if isinstance(_lowerCamelCase,       _lowerCamelCase  ) else eos_token
     __A									= AddedToken(_lowerCamelCase,       lstrip=_lowerCamelCase,       rstrip=_lowerCamelCase  ) if isinstance(_lowerCamelCase,       _lowerCamelCase  ) else sep_token
     __A									= AddedToken(_lowerCamelCase,       lstrip=_lowerCamelCase,       rstrip=_lowerCamelCase  ) if isinstance(_lowerCamelCase,       _lowerCamelCase  ) else cls_token
     __A									= AddedToken(_lowerCamelCase,       lstrip=_lowerCamelCase,       rstrip=_lowerCamelCase  ) if isinstance(_lowerCamelCase,       _lowerCamelCase  ) else unk_token
     __A									= AddedToken(_lowerCamelCase,       lstrip=_lowerCamelCase,       rstrip=_lowerCamelCase  ) if isinstance(_lowerCamelCase,       _lowerCamelCase  ) else pad_token

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

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

     with open(_lowerCamelCase,       encoding='''utf-8'''  ) as vocab_handle:
       __A									= json.load(_lowerCamelCase  )
     __A									= {v: k for k, v in self.encoder.items()}
     __A									= errors  # how to handle errors in decoding
     __A									= bytes_to_unicode()
     __A									= {v: k for k, v in self.byte_encoder.items()}
     with open(_lowerCamelCase,       encoding='''utf-8'''  ) as merges_handle:
       __A									= merges_handle.read().split('''\n'''  )[1:-1]
     __A									= [tuple(merge.split()  ) for merge in bpe_merges]
     __A									= dict(zip(_lowerCamelCase,       range(len(_lowerCamelCase  )  )  )  )
     __A									= {}
     __A									= add_prefix_space

     # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
     __A									= 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 				_SCREAMING_SNAKE_CASE				(	self :     Any  ):

     '''simple docstring'''
     return len(self.encoder  )


   def 				_SCREAMING_SNAKE_CASE				(	self :     Union[str, Any]  ):

     '''simple docstring'''
     return dict(self.encoder,       **self.added_tokens_encoder  )


   def 				_SCREAMING_SNAKE_CASE				(	self :     Optional[Any],       _lowerCamelCase :     List[Any]  ):

     '''simple docstring'''
     if token in self.cache:
       return self.cache[token]
     __A									= tuple(_lowerCamelCase  )
     __A									= get_pairs(_lowerCamelCase  )

     if not pairs:
       return token

     while True:
       __A									= min(_lowerCamelCase,       key=lambda _lowerCamelCase  : self.bpe_ranks.get(_lowerCamelCase,       float('''inf'''  )  )  )
       if bigram not in self.bpe_ranks:
         break
       __A			,    __A									= bigram
       __A									= []
       __A									= 0
       while i < len(_lowerCamelCase  ):
         try:
           __A									= word.index(_lowerCamelCase,       _lowerCamelCase  )
         except ValueError:
           new_word.extend(word[i:]  )
           break
         else:
           new_word.extend(word[i:j]  )
           __A									= j

         if word[i] == first and i < len(_lowerCamelCase  ) - 1 and word[i + 1] == second:
           new_word.append(first + second  )
           i += 2
         else:
           new_word.append(word[i]  )
           i += 1
       __A									= tuple(_lowerCamelCase  )
       __A									= new_word
       if len(_lowerCamelCase  ) == 1:
         break
       else:
         __A									= get_pairs(_lowerCamelCase  )
     __A									= ''' '''.join(_lowerCamelCase  )
     __A									= word
     return word


   def 				_SCREAMING_SNAKE_CASE				(	self :     Optional[int],       _lowerCamelCase :     Dict  ):

     '''simple docstring'''
     __A									= []
     for token in re.findall(self.pat,       _lowerCamelCase  ):
       __A									= ''''''.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(_lowerCamelCase  ).split(''' '''  )  )
     return bpe_tokens


   def 				_SCREAMING_SNAKE_CASE				(	self :     Union[str, Any],       _lowerCamelCase :     Dict  ):

     '''simple docstring'''
     return self.encoder.get(_lowerCamelCase,       self.encoder.get(self.unk_token  )  )


   def 				_SCREAMING_SNAKE_CASE				(	self :     Optional[int],       _lowerCamelCase :     Any  ):

     '''simple docstring'''
     return self.decoder.get(_lowerCamelCase  )


   def 				_SCREAMING_SNAKE_CASE				(	self :     Optional[Any],       _lowerCamelCase :     Dict  ):

     '''simple docstring'''
     __A									= ''''''.join(_lowerCamelCase  )
     __A									= bytearray([self.byte_decoder[c] for c in text]  ).decode('''utf-8''',       errors=self.errors  )
     return text


   def 				_SCREAMING_SNAKE_CASE				(	self :     Dict,       _lowerCamelCase :     str,       _lowerCamelCase :     Optional[str] = None  ):

     '''simple docstring'''
     if not os.path.isdir(_lowerCamelCase  ):
       logger.error(f'Vocabulary path ({save_directory}) should be a directory'  )
       return
     __A									= os.path.join(
         _lowerCamelCase,       (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']  )
     __A									= os.path.join(
         _lowerCamelCase,       (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']  )

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

     __A									= 0
     with open(_lowerCamelCase,       '''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 _lowerCamelCase  : 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									= token_index
         writer.write(''' '''.join(_lowerCamelCase  ) + '''\n'''  )
         index += 1

     return vocab_file, merge_file


   def 				_SCREAMING_SNAKE_CASE				(	self :     Optional[int],       _lowerCamelCase :     List[int],       _lowerCamelCase :     Optional[List[int]] = None,       _lowerCamelCase :     bool = False  ):

     '''simple docstring'''
     if already_has_special_tokens:
       return super().get_special_tokens_mask(
           token_ids_a=_lowerCamelCase,       token_ids_a=_lowerCamelCase,       already_has_special_tokens=_lowerCamelCase  )

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


   def 				_SCREAMING_SNAKE_CASE				(	self :     Any,       _lowerCamelCase :     List[int],       _lowerCamelCase :     Optional[List[int]] = None  ):

     '''simple docstring'''
     __A									= [self.sep_token_id]
     __A									= [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 :     Optional[Any],       _lowerCamelCase :     Union[str, Any],       _lowerCamelCase :     List[str]=False,       **_lowerCamelCase :     List[Any]  ):

     '''simple docstring'''
     __A									= kwargs.pop('''add_prefix_space''',       self.add_prefix_space  )
     if (is_split_into_words or add_prefix_space) and (len(_lowerCamelCase  ) > 0 and not text[0].isspace()):
       __A									= ''' ''' + text
     return (text, kwargs)


   def 				_SCREAMING_SNAKE_CASE				(	self :     Union[str, Any],       _lowerCamelCase :     List[int],       _lowerCamelCase :     Optional[List[int]] = None  ):

     '''simple docstring'''
     return token_ids_a + [self.eos_token_id]






   def 				_SCREAMING_SNAKE_CASE				(	self :     List[Any],       _lowerCamelCase :     "Conversation"  ):

     '''simple docstring'''
     __A									= []
     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(_lowerCamelCase  )

     __A									= '''  '''.join(_lowerCamelCase  )
     __A									= self.encode(_lowerCamelCase  )
     if len(_lowerCamelCase  ) > self.model_max_length:
       __A									= 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




