

import itertools
import random
import unittest

import numpy as np

from transformers import is_speech_available
from transformers.testing_utils import require_torch, require_torchaudio

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
 from transformers import SpeechaTextFeatureExtractor

_UpperCamelCase     			=     random.Random()
def 		UpperCamelCase_(     snake_case__:    Optional[int]    ,       snake_case__:    int=1.0    ,       snake_case__:    Tuple=None    ,       snake_case__:    Any=None  )			->					str:
 if rng is None:
  UpperCAmelCase__     							=							global_rng

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

 return values



@require_torch
@require_torchaudio
class        lowercase		(			unittest.TestCase   ):



 '''simple docstring'''


 def __init__(self			,			__a			,			__a=7			,			__a=400			,			__a=2000			,			__a=24			,			__a=24			,			__a=0.0			,			__a=16000			,			__a=True			,			__a=True			,			)					->   int:






  """simple docstring"""





  UpperCAmelCase__     							=							parent
  UpperCAmelCase__     							=							batch_size
  UpperCAmelCase__     							=							min_seq_length
  UpperCAmelCase__     							=							max_seq_length
  UpperCAmelCase__     							=							(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
  UpperCAmelCase__     							=							feature_size
  UpperCAmelCase__     							=							num_mel_bins
  UpperCAmelCase__     							=							padding_value
  UpperCAmelCase__     							=							sampling_rate
  UpperCAmelCase__     							=							return_attention_mask
  UpperCAmelCase__     							=							do_normalize




 def     UpperCamelCase__				(self					)					->   Tuple:






  """simple docstring"""





  return {
      "feature_size": self.feature_size,
      "num_mel_bins": self.num_mel_bins,
      "padding_value": self.padding_value,
      "sampling_rate": self.sampling_rate,
      "return_attention_mask": self.return_attention_mask,
      "do_normalize": self.do_normalize,
  }




 def     UpperCamelCase__				(self			,			__a=False			,			__a=False					)					->   Tuple:






  """simple docstring"""





  def _flatten(__a					):
   return list(itertools.chain(*__a					)					)

  if equal_length:
   UpperCAmelCase__     							=							[floats_list((self.max_seq_length, self.feature_size)					) for _ in range(self.batch_size					)]
  else:
   # make sure that inputs increase in size
   UpperCAmelCase__     							=							[
       floats_list((x, self.feature_size)					)
       for x in range(self.min_seq_length			,			self.max_seq_length			,			self.seq_length_diff					)
   ]
  if numpify:
   UpperCAmelCase__     							=							[np.asarray(__a					) for x in speech_inputs]
  return speech_inputs







@require_torch
@require_torchaudio
class        lowercase		(			_lowercase					,							unittest.TestCase   ):



 '''simple docstring'''


 __SCREAMING_SNAKE_CASE   				=							SpeechaTextFeatureExtractor if is_speech_available() else None




 def     UpperCamelCase__				(self					)					->   int:






  """simple docstring"""





  UpperCAmelCase__     							=							SpeechaTextFeatureExtractionTester(self					)




 def     UpperCamelCase__				(self			,			__a					)					->   str:






  """simple docstring"""





  self.assertTrue(np.all(np.mean(__a			,			axis=0					) < 1E-3					)					)
  self.assertTrue(np.all(np.abs(np.var(__a			,			axis=0					) - 1					) < 1E-3					)					)




 def     UpperCamelCase__				(self					)					->   List[Any]:






  """simple docstring"""





  UpperCAmelCase__     							=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
  # create three inputs of length 800, 1000, and 1200
  UpperCAmelCase__     							=							[floats_list((1, x)					)[0] for x in range(800			,			1400			,			200					)]
  UpperCAmelCase__     							=							[np.asarray(__a					) for speech_input in speech_inputs]

  # Test feature size
  UpperCAmelCase__     							=							feature_extractor(__a			,			padding=__a			,			return_tensors='np'					).input_features
  self.assertTrue(input_features.ndim == 3					)
  self.assertTrue(input_features.shape[-1] == feature_extractor.feature_size					)

  # Test not batched input
  UpperCAmelCase__     							=							feature_extractor(speech_inputs[0]			,			return_tensors='np'					).input_features
  UpperCAmelCase__     							=							feature_extractor(np_speech_inputs[0]			,			return_tensors='np'					).input_features
  self.assertTrue(np.allclose(__a			,			__a			,			atol=1E-3					)					)

  # Test batched
  UpperCAmelCase__     							=							feature_extractor(__a			,			return_tensors='np'					).input_features
  UpperCAmelCase__     							=							feature_extractor(__a			,			return_tensors='np'					).input_features
  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.
  UpperCAmelCase__     							=							[floats_list((1, x)					)[0] for x in (800, 800, 800)]
  UpperCAmelCase__     							=							np.asarray(__a					)
  UpperCAmelCase__     							=							feature_extractor(__a			,			return_tensors='np'					).input_features
  UpperCAmelCase__     							=							feature_extractor(__a			,			return_tensors='np'					).input_features
  for enc_seq_a, enc_seq_a in zip(__a			,			__a					):
   self.assertTrue(np.allclose(__a			,			__a			,			atol=1E-3					)					)




 def     UpperCamelCase__				(self					)					->   Optional[Any]:






  """simple docstring"""





  UpperCAmelCase__     							=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
  UpperCAmelCase__     							=							[floats_list((1, x)					)[0] for x in range(800			,			1400			,			200					)]

  UpperCAmelCase__     							=							['longest', 'max_length', 'do_not_pad']
  UpperCAmelCase__     							=							[None, 16, None]
  for max_length, padding in zip(__a			,			__a					):
   UpperCAmelCase__     							=							feature_extractor(
       __a			,			padding=__a			,			max_length=__a			,			return_attention_mask=__a					)
   UpperCAmelCase__     							=							inputs.input_features
   UpperCAmelCase__     							=							inputs.attention_mask
   UpperCAmelCase__     							=							[np.sum(__a					) for x in attention_mask]

   self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]					)
   self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]					)
   self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]					)




 def     UpperCamelCase__				(self					)					->   Optional[Any]:






  """simple docstring"""





  UpperCAmelCase__     							=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
  UpperCAmelCase__     							=							[floats_list((1, x)					)[0] for x in range(800			,			1400			,			200					)]

  UpperCAmelCase__     							=							['longest', 'max_length', 'do_not_pad']
  UpperCAmelCase__     							=							[None, 16, None]
  for max_length, padding in zip(__a			,			__a					):
   UpperCAmelCase__     							=							feature_extractor(
       __a			,			max_length=__a			,			padding=__a			,			return_tensors='np'			,			return_attention_mask=__a					)
   UpperCAmelCase__     							=							inputs.input_features
   UpperCAmelCase__     							=							inputs.attention_mask
   UpperCAmelCase__     							=							[np.sum(__a					) for x in attention_mask]

   self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]					)
   self.assertTrue(input_features[0][fbank_feat_lengths[0] :].sum() < 1E-6					)
   self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]					)
   self.assertTrue(input_features[0][fbank_feat_lengths[1] :].sum() < 1E-6					)
   self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]					)




 def     UpperCamelCase__				(self					)					->   List[Any]:






  """simple docstring"""





  UpperCAmelCase__     							=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
  UpperCAmelCase__     							=							[floats_list((1, x)					)[0] for x in range(800			,			1400			,			200					)]
  UpperCAmelCase__     							=							feature_extractor(
      __a			,			padding='max_length'			,			max_length=4			,			truncation=__a			,			return_tensors='np'			,			return_attention_mask=__a			,			)
  UpperCAmelCase__     							=							inputs.input_features
  UpperCAmelCase__     							=							inputs.attention_mask
  UpperCAmelCase__     							=							np.sum(attention_mask == 1			,			axis=1					)

  self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]					)
  self._check_zero_mean_unit_variance(input_features[1]					)
  self._check_zero_mean_unit_variance(input_features[2]					)




 def     UpperCamelCase__				(self					)					->   Tuple:






  """simple docstring"""





  UpperCAmelCase__     							=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
  UpperCAmelCase__     							=							[floats_list((1, x)					)[0] for x in range(800			,			1400			,			200					)]
  UpperCAmelCase__     							=							feature_extractor(
      __a			,			padding='longest'			,			max_length=4			,			truncation=__a			,			return_tensors='np'			,			return_attention_mask=__a			,			)
  UpperCAmelCase__     							=							inputs.input_features
  UpperCAmelCase__     							=							inputs.attention_mask
  UpperCAmelCase__     							=							np.sum(attention_mask == 1			,			axis=1					)

  self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]					)
  self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]					)
  self._check_zero_mean_unit_variance(input_features[2]					)

  # make sure that if max_length < longest -> then pad to max_length
  self.assertEqual(input_features.shape			,			(3, 4, 24)					)

  UpperCAmelCase__     							=							[floats_list((1, x)					)[0] for x in range(800			,			1400			,			200					)]
  UpperCAmelCase__     							=							feature_extractor(
      __a			,			padding='longest'			,			max_length=16			,			truncation=__a			,			return_tensors='np'			,			return_attention_mask=__a			,			)
  UpperCAmelCase__     							=							inputs.input_features
  UpperCAmelCase__     							=							inputs.attention_mask
  UpperCAmelCase__     							=							np.sum(attention_mask == 1			,			axis=1					)

  self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]					)
  self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]					)
  self._check_zero_mean_unit_variance(input_features[2]					)

  # make sure that if max_length < longest -> then pad to max_length
  self.assertEqual(input_features.shape			,			(3, 6, 24)					)




 def     UpperCamelCase__				(self					)					->   List[str]:






  """simple docstring"""





  import torch

  UpperCAmelCase__     							=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
  UpperCAmelCase__     							=							np.random.rand(100			,			32					).astype(np.floataa					)
  UpperCAmelCase__     							=							np_speech_inputs.tolist()

  for inputs in [py_speech_inputs, np_speech_inputs]:
   UpperCAmelCase__     							=							feature_extractor.pad([{'input_features': inputs}]			,			return_tensors='np'					)
   self.assertTrue(np_processed.input_features.dtype == np.floataa					)
   UpperCAmelCase__     							=							feature_extractor.pad([{'input_features': inputs}]			,			return_tensors='pt'					)
   self.assertTrue(pt_processed.input_features.dtype == torch.floataa					)




 def     UpperCamelCase__				(self			,			__a					)					->   List[Any]:






  """simple docstring"""





  from datasets import load_dataset

  UpperCAmelCase__     							=							load_dataset('hf-internal-testing/librispeech_asr_dummy'			,			'clean'			,			split='validation'					)
  # automatic decoding with librispeech
  UpperCAmelCase__     							=							ds.sort('id'					).select(range(__a					)					)[:num_samples]['audio']

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




 def     UpperCamelCase__				(self					)					->   Dict:






  """simple docstring"""





  UpperCAmelCase__     							=							np.array([
      -1.57_45, -1.77_13, -1.70_20, -1.60_69, -1.22_50, -1.11_05, -0.90_72, -0.82_41,
      -1.23_10, -0.80_98, -0.33_20, -0.41_01, -0.79_85, -0.49_96, -0.82_13, -0.91_28,
      -1.04_20, -1.12_86, -1.04_40, -0.79_99, -0.84_05, -1.22_75, -1.54_43, -1.46_25,
  ]					)
  # fmt: on

  UpperCAmelCase__     							=							self._load_datasamples(1					)
  UpperCAmelCase__     							=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
  UpperCAmelCase__     							=							feature_extractor(__a			,			return_tensors='pt'					).input_features
  self.assertEquals(input_features.shape			,			(1, 584, 24)					)
  self.assertTrue(np.allclose(input_features[0, 0, :30]			,			__a			,			atol=1E-4					)					)








import json
import os
from typing import Dict, List, Optional, Tuple

import regex as re

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


a_      :      Dict    =     logging.get_logger(__name__)


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

a_      :      Union[str, Any]    =     {
    'vocab_file': {
        'facebook/blenderbot_small-90M': 'https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/vocab.json'
    },
    'merges_file': {
        'facebook/blenderbot_small-90M': 'https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/merges.txt'
    },
    'tokenizer_config_file': {
        'facebook/blenderbot_small-90M': (
            'https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/tokenizer_config.json'
        )
    },
}

a_      :      List[Any]    =     {'facebook/blenderbot_small-90M': 5_12}






def     __a					(		__UpperCAmelCase       ):
 a__												= set()
 a__												= word[0]
 for char in word[1:]:
  pairs.add((prev_char, char)       )
  a__												= char

 a__												= set(__UpperCAmelCase       )
 return pairs

class    __UpperCamelCase			(       _lowercase				):

 """simple docstring"""



 _lowercase   :    Union[str, Any]      						=					VOCAB_FILES_NAMES
 _lowercase   :    List[Any]      						=					PRETRAINED_VOCAB_FILES_MAP
 _lowercase   :    Tuple      						=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _lowercase   :    int      						=					['''input_ids''', '''attention_mask''']

 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE="__start__"		,				SCREAMING_SNAKE_CASE="__end__"		,				SCREAMING_SNAKE_CASE="__unk__"		,				SCREAMING_SNAKE_CASE="__null__"		,				**SCREAMING_SNAKE_CASE		,				) ->			List[str]:
  super().__init__(unk_token=SCREAMING_SNAKE_CASE		,				bos_token=SCREAMING_SNAKE_CASE		,				eos_token=SCREAMING_SNAKE_CASE		,				pad_token=SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      )

  with open(SCREAMING_SNAKE_CASE		,				encoding='''utf-8'''      ) as vocab_handle:
   a__												= json.load(SCREAMING_SNAKE_CASE      )
  a__												= {v: k for k, v in self.encoder.items()}
  with open(SCREAMING_SNAKE_CASE		,				encoding='''utf-8'''      ) as merges_handle:
   a__												= merges_handle.read().split('''\n'''      )[1:-1]
  a__												= [tuple(merge.split()      ) for merge in merges]
  a__												= dict(zip(SCREAMING_SNAKE_CASE		,				range(len(SCREAMING_SNAKE_CASE      )      )      )      )
  a__												= {}

 @property
 def 						_UpperCAmelCase       (	self      ) ->			int:
  return len(self.encoder      )

 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  return dict(self.encoder		,				**self.added_tokens_encoder      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			str:
  if token in self.cache:
   return self.cache[token]
  a__												= re.sub('''([.,!?()])'''		,				R''' \1'''		,				SCREAMING_SNAKE_CASE      )
  a__												= re.sub('''(\')'''		,				R''' \1 '''		,				SCREAMING_SNAKE_CASE      )
  a__												= re.sub(R'''\s{2,}'''		,				''' '''		,				SCREAMING_SNAKE_CASE      )
  if "\n" in token:
   a__												= token.replace('''\n'''		,				''' __newln__'''      )

  a__												= token.split(''' '''      )
  a__												= []
  for token in tokens:
   if not len(SCREAMING_SNAKE_CASE      ):
    continue

   a__												= token.lower()
   a__												= tuple(SCREAMING_SNAKE_CASE      )
   a__												= tuple(list(word[:-1]      ) + [word[-1] + '''</w>''']      )
   a__												= get_pairs(SCREAMING_SNAKE_CASE      )

   if not pairs:
    words.append(SCREAMING_SNAKE_CASE      )
    continue

   while True:
    a__												= min(SCREAMING_SNAKE_CASE		,				key=lambda SCREAMING_SNAKE_CASE      : self.bpe_ranks.get(SCREAMING_SNAKE_CASE		,				float('''inf'''      )      )      )
    if bigram not in self.bpe_ranks:
     break
    a__							,			a__												= bigram
    a__												= []
    a__												= 0

    while i < len(SCREAMING_SNAKE_CASE      ):
     try:
      a__												= word.index(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )
      new_word.extend(word[i:j]      )
      a__												= j
     except ValueError:
      new_word.extend(word[i:]      )
      break

     if word[i] == first and i < len(SCREAMING_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(SCREAMING_SNAKE_CASE      )
    a__												= new_word
    if len(SCREAMING_SNAKE_CASE      ) == 1:
     break
    else:
     a__												= get_pairs(SCREAMING_SNAKE_CASE      )
   a__												= '''@@ '''.join(SCREAMING_SNAKE_CASE      )
   a__												= word[:-4]

   a__												= word
   words.append(SCREAMING_SNAKE_CASE      )
  return " ".join(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			List[str]:
  a__												= []

  a__												= re.findall(R'''\S+\n?'''		,				SCREAMING_SNAKE_CASE      )

  for token in words:
   split_tokens.extend(list(self.bpe(SCREAMING_SNAKE_CASE      ).split(''' '''      )      )      )
  return split_tokens

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			int:
  a__												= token.lower()
  return self.encoder.get(SCREAMING_SNAKE_CASE		,				self.encoder.get(self.unk_token      )      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			str:
  return self.decoder.get(SCREAMING_SNAKE_CASE		,				self.unk_token      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			str:
  a__												= ''' '''.join(SCREAMING_SNAKE_CASE      ).replace('''@@ '''		,				''''''      ).strip()
  return out_string



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = None      ) ->			Tuple[str]:
  if not os.path.isdir(SCREAMING_SNAKE_CASE      ):
   logger.error(f"Vocabulary path ({save_directory}) should be a directory"      )
   return
  a__												= os.path.join(
      SCREAMING_SNAKE_CASE		,				(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']      )
  a__												= os.path.join(
      SCREAMING_SNAKE_CASE		,				(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']      )

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

  a__												= 0
  with open(SCREAMING_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 SCREAMING_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__												= token_index
    writer.write(''' '''.join(SCREAMING_SNAKE_CASE      ) + '''\n'''      )
    index += 1

  return vocab_file, merge_file





