
'''simple docstring'''






import itertools
import random
import unittest

import numpy as np

from transformers import ASTFeatureExtractor
from transformers.testing_utils import require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


UpperCAmelCase_       :   str   =       random.Random()

if is_torch_available():
 import torch







def 		A_       (	_lowerCAmelCase   :    Optional[int]				,	_lowerCAmelCase   :    int=1.0				,	_lowerCAmelCase   :    str=None				,	_lowerCAmelCase   :    Tuple=None							):



      """simple docstring"""




      if rng is None:
            _lowerCamelCase				:   List[str]								=   global_rng

      _lowerCamelCase				:   Dict								=   []
      for batch_idx in range(shape[0]							):
            values.append([]							)
            for _ in range(shape[1]							):
                  values[-1].append(rng.random() * scale							)

      return values

class  UpperCAmelCase__						(       unittest.TestCase ):



    def __init__(	self      :       Optional[int],__A      :       Tuple,__A      :       List[str]=7,__A      :       Any=4_0_0,__A      :       Tuple=2_0_0_0,__A      :       Union[str, Any]=1,__A      :       Union[str, Any]=0.0,__A      :       Union[str, Any]=1_6_0_0_0,__A      :       int=True,__A      :       Optional[int]=True,):
          _lowerCamelCase				:   List[str]								=   parent
          _lowerCamelCase				:   Tuple								=   batch_size
          _lowerCamelCase				:   Dict								=   min_seq_length
          _lowerCamelCase				:   Optional[int]								=   max_seq_length
          _lowerCamelCase				:   List[Any]								=   (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
          _lowerCamelCase				:   str								=   feature_size
          _lowerCamelCase				:   List[str]								=   padding_value
          _lowerCamelCase				:   List[str]								=   sampling_rate
          _lowerCamelCase				:   Optional[Any]								=   return_attention_mask
          _lowerCamelCase				:   Any								=   do_normalize



    def   lowerCamelCase_  (	self      :       Any     ):
          return {
              "feature_size": self.feature_size,
              "padding_value": self.padding_value,
              "sampling_rate": self.sampling_rate,
              "return_attention_mask": self.return_attention_mask,
              "do_normalize": self.do_normalize,
          }



    def   lowerCamelCase_  (	self      :       Any,__A      :       str=False,__A      :       Any=False     ):
          def _flatten(__A      :       Union[str, Any]     ):
                return list(itertools.chain(*__A     )     )

          if equal_length:
                _lowerCamelCase				:   Optional[Any]								=   floats_list((self.batch_size, self.max_seq_length)     )
          else:
                # make sure that inputs increase in size
                _lowerCamelCase				:   Optional[int]								=   [
                    _flatten(floats_list((x, self.feature_size)     )     )
                    for x in range(self.min_seq_length,self.max_seq_length,self.seq_length_diff     )
                ]

          if numpify:
                _lowerCamelCase				:   Optional[int]								=   [np.asarray(__A     ) for x in speech_inputs]

          return speech_inputs


@require_torch
@require_torchaudio
class  UpperCAmelCase__						(       A						,	unittest.TestCase ):
    lowerCAmelCase_				    =		ASTFeatureExtractor



    def   lowerCamelCase_  (	self      :       List[str]     ):
          _lowerCamelCase				:   List[str]								=   ASTFeatureExtractionTester(self     )



    def   lowerCamelCase_  (	self      :       Tuple     ):
          # Tests that all call wrap to encode_plus and batch_encode_plus
          _lowerCamelCase				:   List[str]								=   self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()     )
          # create three inputs of length 800, 1000, and 1200
          _lowerCamelCase				:   Union[str, Any]								=   [floats_list((1, x)     )[0] for x in range(8_0_0,1_4_0_0,2_0_0     )]
          _lowerCamelCase				:   str								=   [np.asarray(__A     ) for speech_input in speech_inputs]

          # Test not batched input
          _lowerCamelCase				:   Any								=   feat_extract(speech_inputs[0],return_tensors="np"     ).input_values
          _lowerCamelCase				:   List[Any]								=   feat_extract(np_speech_inputs[0],return_tensors="np"     ).input_values
          self.assertTrue(np.allclose(__A,__A,atol=1e-3     )     )

          # Test batched
          _lowerCamelCase				:   Dict								=   feat_extract(__A,padding=__A,return_tensors="np"     ).input_values
          _lowerCamelCase				:   str								=   feat_extract(__A,padding=__A,return_tensors="np"     ).input_values
          for enc_seq_a, enc_seq_a in zip(__A,__A     ):
                self.assertTrue(np.allclose(__A,__A,atol=1e-3     )     )

          # Test 2-D numpy arrays are batched.
          _lowerCamelCase				:   Dict								=   [floats_list((1, x)     )[0] for x in (8_0_0, 8_0_0, 8_0_0)]
          _lowerCamelCase				:   int								=   np.asarray(__A     )
          _lowerCamelCase				:   Dict								=   feat_extract(__A,return_tensors="np"     ).input_values
          _lowerCamelCase				:   Optional[int]								=   feat_extract(__A,return_tensors="np"     ).input_values
          for enc_seq_a, enc_seq_a in zip(__A,__A     ):
                self.assertTrue(np.allclose(__A,__A,atol=1e-3     )     )



    @require_torch
    def   lowerCamelCase_  (	self      :       int     ):
          import torch

          _lowerCamelCase				:   List[str]								=   self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()     )
          _lowerCamelCase				:   List[str]								=   np.random.rand(1_0_0     ).astype(np.floataa     )
          _lowerCamelCase				:   Optional[int]								=   np_speech_inputs.tolist()

          for inputs in [py_speech_inputs, np_speech_inputs]:
                _lowerCamelCase				:   int								=   feature_extractor.pad([{"input_values": inputs}],return_tensors="np"     )
                self.assertTrue(np_processed.input_values.dtype == np.floataa     )
                _lowerCamelCase				:   Tuple								=   feature_extractor.pad([{"input_values": inputs}],return_tensors="pt"     )
                self.assertTrue(pt_processed.input_values.dtype == torch.floataa     )



    def   lowerCamelCase_  (	self      :       Optional[int],__A      :       Optional[int]     ):
          from datasets import load_dataset

          _lowerCamelCase				:   Optional[Any]								=   load_dataset("hf-internal-testing/librispeech_asr_dummy","clean",split="validation"     )
          # automatic decoding with librispeech
          _lowerCamelCase				:   List[Any]								=   ds.sort("id"     ).select(range(__A     )     )[:num_samples]["audio"]

          return [x["array"] for x in speech_samples]



    @require_torch
    def   lowerCamelCase_  (	self      :       List[str]     ):
          # fmt: off
          _lowerCamelCase				:   Optional[int]								=   torch.tensor(
              [-0.9894, -1.2776, -0.9066, -1.2776, -0.9349, -1.2609, -1.0386, -1.2776,
               -1.1561, -1.2776, -1.2052, -1.2723, -1.2190, -1.2132, -1.2776, -1.1133,
               -1.1953, -1.1343, -1.1584, -1.2203, -1.1770, -1.2474, -1.2381, -1.1936,
               -0.9270, -0.8317, -0.8049, -0.7706, -0.7565, -0.7869]     )
          # fmt: on

          _lowerCamelCase				:   Any								=   self._load_datasamples(1     )
          _lowerCamelCase				:   int								=   ASTFeatureExtractor()
          _lowerCamelCase				:   str								=   feature_extractor(__A,return_tensors="pt"     ).input_values
          self.assertEquals(input_values.shape,(1, 1_0_2_4, 1_2_8)     )
          self.assertTrue(torch.allclose(input_values[0, 0, :3_0],__A,atol=1e-4     )     )

'''simple docstring'''






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

UpperCAmelCase_       :   Any   =       logging.get_logger(__name__)


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

UpperCAmelCase_       :   List[str]   =       {
    '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'
    },
}

UpperCAmelCase_       :   List[Any]   =       {'facebook/blenderbot-3B': 128}

class  UpperCAmelCase__						(       A ):
    lowerCAmelCase_				    =		VOCAB_FILES_NAMES
    lowerCAmelCase_				    =		PRETRAINED_VOCAB_FILES_MAP
    lowerCAmelCase_				    =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    lowerCAmelCase_				    =		['input_ids', 'attention_mask']
    lowerCAmelCase_				    =		BlenderbotTokenizer



    def __init__(	self      :       Dict,__A      :       Optional[Any]=None,__A      :       List[str]=None,__A      :       Optional[Any]=None,__A      :       List[Any]="replace",__A      :       List[Any]="<s>",__A      :       str="</s>",__A      :       List[str]="</s>",__A      :       List[Any]="<s>",__A      :       Union[str, Any]="<unk>",__A      :       Optional[Any]="<pad>",__A      :       Dict="<mask>",__A      :       Any=False,__A      :       Tuple=True,**__A      :       Dict,):
          super().__init__(
              __A,__A,tokenizer_file=__A,errors=__A,bos_token=__A,eos_token=__A,sep_token=__A,cls_token=__A,unk_token=__A,pad_token=__A,mask_token=__A,add_prefix_space=__A,trim_offsets=__A,**__A,)

          _lowerCamelCase				:   Tuple								=   json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()     )
          if pre_tok_state.get("add_prefix_space",__A     ) != add_prefix_space:
                _lowerCamelCase				:   Optional[Any]								=   getattr(__A,pre_tok_state.pop("type"     )     )
                _lowerCamelCase				:   List[Any]								=   add_prefix_space
                _lowerCamelCase				:   Optional[Any]								=   pre_tok_class(**__A     )

          _lowerCamelCase				:   Any								=   add_prefix_space

          _lowerCamelCase				:   Any								=   "post_processor"
          _lowerCamelCase				:   Optional[Any]								=   getattr(self.backend_tokenizer,__A,__A     )
          if tokenizer_component_instance:
                _lowerCamelCase				:   Tuple								=   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:
                      _lowerCamelCase				:   List[Any]								=   tuple(state["sep"]     )
                if "cls" in state:
                      _lowerCamelCase				:   Optional[int]								=   tuple(state["cls"]     )

                _lowerCamelCase				:   List[str]								=   False

                if state.get("add_prefix_space",__A     ) != add_prefix_space:
                      _lowerCamelCase				:   List[str]								=   add_prefix_space
                      _lowerCamelCase				:   int								=   True

                if state.get("trim_offsets",__A     ) != trim_offsets:
                      _lowerCamelCase				:   List[str]								=   trim_offsets
                      _lowerCamelCase				:   Dict								=   True

                if changes_to_apply:
                      _lowerCamelCase				:   Tuple								=   getattr(__A,state.pop("type"     )     )
                      _lowerCamelCase				:   Union[str, Any]								=   component_class(**__A     )
                      setattr(self.backend_tokenizer,__A,__A     )



    @property
    # Copied from transformers.models.roberta.tokenization_roberta_fast.RobertaTokenizerFast.mask_token with Roberta->Blenderbot, RoBERTa->Blenderbot
    def   lowerCamelCase_  (	self      :       Optional[int]     ):
          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   lowerCamelCase_  (	self      :       Optional[Any],__A      :       List[Any]     ):
          _lowerCamelCase				:   Any								=   AddedToken(__A,lstrip=__A,rstrip=__A     ) if isinstance(__A,__A     ) else value
          _lowerCamelCase				:   List[str]								=   value



    def   lowerCamelCase_  (	self      :       Optional[int],*__A      :       Optional[Any],**__A      :       List[str]     ):
          _lowerCamelCase				:   int								=   kwargs.get("is_split_into_words",__A     )
          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(*__A,**__A     )



    def   lowerCamelCase_  (	self      :       str,*__A      :       Union[str, Any],**__A      :       List[Any]     ):
          _lowerCamelCase				:   Optional[int]								=   kwargs.get("is_split_into_words",__A     )

          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(*__A,**__A     )



    def   lowerCamelCase_  (	self      :       str,__A      :       str,__A      :       Optional[str] = None     ):
          _lowerCamelCase				:   Optional[int]								=   self._tokenizer.model.save(__A,name=__A     )
          return tuple(__A     )



    def   lowerCamelCase_  (	self      :       Optional[int],__A      :       List[int],__A      :       Optional[List[int]] = None     ):
          _lowerCamelCase				:   int								=   [self.sep_token_id]
          _lowerCamelCase				:   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   lowerCamelCase_  (	self      :       Dict,__A      :       List[int],__A      :       Optional[List[int]] = None     ):
          return token_ids_a + [self.eos_token_id]



    def   lowerCamelCase_  (	self      :       List[str],__A      :       "Conversation"     ):
          _lowerCamelCase				:   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(__A     )

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