


'''simple docstring'''
import itertools
import random
import unittest

import numpy as np

from transformers import WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST, WavaVecaConfig, WavaVecaFeatureExtractor
from transformers.testing_utils import require_torch, slow

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


UpperCamelCase__    					=      random.Random()





def 					a__       (					lowerCAmelCase__  ,					lowerCAmelCase__=1.0  ,					lowerCAmelCase__=None  ,					lowerCAmelCase__=None						)    ->	List[str]:
    if rng is None:
        UpperCAmelCase__			:		str													=				global_rng

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

    return values




class 			lowerCamelCase_   (	unittest.TestCase      ):






      def __init__(					self				:    List[Any]						,						_A				:    Tuple						,						_A				:    List[str]=7						,						_A				:    Any=400						,						_A				:    Tuple=2_000						,						_A				:    Union[str, Any]=1						,						_A				:    Any=0.0						,						_A				:    List[str]=16_000						,						_A				:    List[Any]=True						,						_A				:    Tuple=True						,						):


          '''simple docstring'''


          UpperCAmelCase__			:		Optional[Any]													=				parent
          UpperCAmelCase__			:		Tuple													=				batch_size
          UpperCAmelCase__			:		str													=				min_seq_length
          UpperCAmelCase__			:		Optional[int]													=				max_seq_length
          UpperCAmelCase__			:		Optional[Any]													=				(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
          UpperCAmelCase__			:		Dict													=				feature_size
          UpperCAmelCase__			:		Any													=				padding_value
          UpperCAmelCase__			:		int													=				sampling_rate
          UpperCAmelCase__			:		str													=				return_attention_mask
          UpperCAmelCase__			:		Optional[int]													=				do_normalize






      def lowercase_							(					self				:    List[Any] ):


          '''simple docstring'''


          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 lowercase_							(					self				:    int						,						_A				:    Optional[int]=False						,						_A				:    int=False ):


          '''simple docstring'''


          def _flatten(_A				:    Optional[Any] ):
              return list(itertools.chain(*_A ) )

          if equal_length:
              UpperCAmelCase__			:		List[str]													=				floats_list((self.batch_size, self.max_seq_length) )
          else:
              # make sure that inputs increase in size
              UpperCAmelCase__			:		Any													=				[
                  _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:
              UpperCAmelCase__			:		List[str]													=				[np.asarray(_A ) for x in speech_inputs]

          return speech_inputs




class 			lowerCamelCase_   (	__a     ,       unittest.TestCase      ):
      lowerCAmelCase__		     =     WavaVecaFeatureExtractor






      def lowercase_							(					self				:    Tuple ):


          '''simple docstring'''


          UpperCAmelCase__			:		Optional[int]													=				WavaVecaFeatureExtractionTester(self )






      def lowercase_							(					self				:    Dict						,						_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 lowercase_							(					self				:    int ):


          '''simple docstring'''


          UpperCAmelCase__			:		Optional[int]													=				self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
          # create three inputs of length 800, 1000, and 1200
          UpperCAmelCase__			:		Union[str, Any]													=				[floats_list((1, x) )[0] for x in range(800						,						1_400						,						200 )]
          UpperCAmelCase__			:		Any													=				[np.asarray(_A ) for speech_input in speech_inputs]

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

          # Test batched
          UpperCAmelCase__			:		Optional[int]													=				feat_extract(_A						,						return_tensors='''np''' ).input_values
          UpperCAmelCase__			:		Dict													=				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 ) )

          # Test 2-D numpy arrays are batched.
          UpperCAmelCase__			:		str													=				[floats_list((1, x) )[0] for x in (800, 800, 800)]
          UpperCAmelCase__			:		int													=				np.asarray(_A )
          UpperCAmelCase__			:		str													=				feat_extract(_A						,						return_tensors='''np''' ).input_values
          UpperCAmelCase__			:		List[str]													=				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 ) )






      def lowercase_							(					self				:    Optional[Any] ):


          '''simple docstring'''


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

          UpperCAmelCase__			:		str													=				['''longest''', '''max_length''', '''do_not_pad''']
          UpperCAmelCase__			:		str													=				[None, 1_600, None]
          for max_length, padding in zip(_A						,						_A ):
              UpperCAmelCase__			:		Any													=				feat_extract(_A						,						padding=_A						,						max_length=_A						,						return_tensors='''np''' )
              UpperCAmelCase__			:		Optional[int]													=				processed.input_values

              self._check_zero_mean_unit_variance(input_values[0][:800] )
              self.assertTrue(input_values[0][800:].sum() < 1e-6 )
              self._check_zero_mean_unit_variance(input_values[1][:1_000] )
              self.assertTrue(input_values[0][1_000:].sum() < 1e-6 )
              self._check_zero_mean_unit_variance(input_values[2][:1_200] )






      def lowercase_							(					self				:    List[Any] ):


          '''simple docstring'''


          UpperCAmelCase__			:		Dict													=				self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
          UpperCAmelCase__			:		Tuple													=				range(800						,						1_400						,						200 )
          UpperCAmelCase__			:		Tuple													=				[floats_list((1, x) )[0] for x in lengths]

          UpperCAmelCase__			:		Any													=				['''longest''', '''max_length''', '''do_not_pad''']
          UpperCAmelCase__			:		Optional[Any]													=				[None, 1_600, None]

          for max_length, padding in zip(_A						,						_A ):
              UpperCAmelCase__			:		List[Any]													=				feat_extract(_A						,						max_length=_A						,						padding=_A )
              UpperCAmelCase__			:		Any													=				processed.input_values

              self._check_zero_mean_unit_variance(input_values[0][:800] )
              self._check_zero_mean_unit_variance(input_values[1][:1_000] )
              self._check_zero_mean_unit_variance(input_values[2][:1_200] )






      def lowercase_							(					self				:    Any ):


          '''simple docstring'''


          UpperCAmelCase__			:		Optional[Any]													=				self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
          UpperCAmelCase__			:		Any													=				[floats_list((1, x) )[0] for x in range(800						,						1_400						,						200 )]
          UpperCAmelCase__			:		Optional[Any]													=				feat_extract(
              _A						,						truncation=_A						,						max_length=1_000						,						padding='''max_length'''						,						return_tensors='''np''' )
          UpperCAmelCase__			:		str													=				processed.input_values

          self._check_zero_mean_unit_variance(input_values[0, :800] )
          self._check_zero_mean_unit_variance(input_values[1] )
          self._check_zero_mean_unit_variance(input_values[2] )






      def lowercase_							(					self				:    Dict ):


          '''simple docstring'''


          UpperCAmelCase__			:		List[str]													=				self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
          UpperCAmelCase__			:		List[Any]													=				[floats_list((1, x) )[0] for x in range(800						,						1_400						,						200 )]
          UpperCAmelCase__			:		Dict													=				feat_extract(
              _A						,						truncation=_A						,						max_length=1_000						,						padding='''longest'''						,						return_tensors='''np''' )
          UpperCAmelCase__			:		Optional[int]													=				processed.input_values

          self._check_zero_mean_unit_variance(input_values[0, :800] )
          self._check_zero_mean_unit_variance(input_values[1, :1_000] )
          self._check_zero_mean_unit_variance(input_values[2] )

          # make sure that if max_length < longest -> then pad to max_length
          self.assertTrue(input_values.shape == (3, 1_000) )

          UpperCAmelCase__			:		List[Any]													=				[floats_list((1, x) )[0] for x in range(800						,						1_400						,						200 )]
          UpperCAmelCase__			:		Tuple													=				feat_extract(
              _A						,						truncation=_A						,						max_length=2_000						,						padding='''longest'''						,						return_tensors='''np''' )
          UpperCAmelCase__			:		Union[str, Any]													=				processed.input_values

          self._check_zero_mean_unit_variance(input_values[0, :800] )
          self._check_zero_mean_unit_variance(input_values[1, :1_000] )
          self._check_zero_mean_unit_variance(input_values[2] )

          # make sure that if max_length > longest -> then pad to longest
          self.assertTrue(input_values.shape == (3, 1_200) )






      @require_torch
      def lowercase_							(					self				:    Optional[Any] ):


          '''simple docstring'''


          import torch

          UpperCAmelCase__			:		int													=				self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
          UpperCAmelCase__			:		List[Any]													=				np.random.rand(100 ).astype(np.floataa )
          UpperCAmelCase__			:		Dict													=				np_speech_inputs.tolist()

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






      @slow
      @require_torch
      def lowercase_							(					self				:    List[Any] ):


          '''simple docstring'''


          for model_id in WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST:
              UpperCAmelCase__			:		Any													=				WavaVecaConfig.from_pretrained(_A )
              UpperCAmelCase__			:		List[Any]													=				WavaVecaFeatureExtractor.from_pretrained(_A )

              # only "layer" feature extraction norm should make use of
              # attention_mask
              self.assertEqual(feat_extract.return_attention_mask						,						config.feat_extract_norm == '''layer''' )




'''simple docstring'''
import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


UpperCamelCase__    					=      logging.get_logger(__name__)

UpperCamelCase__    					=      '''▁'''

UpperCamelCase__    					=      {'''vocab_file''': '''sentencepiece.bpe.model'''}

UpperCamelCase__    					=      {
    '''vocab_file''': {
        '''facebook/xglm-564M''': '''https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model''',
    }
}

UpperCamelCase__    					=      {
    '''facebook/xglm-564M''': 2_0_4_8,
}




class 			lowerCamelCase_   (	__a      ):
      lowerCAmelCase__		     =     VOCAB_FILES_NAMES
      lowerCAmelCase__		     =     PRETRAINED_VOCAB_FILES_MAP
      lowerCAmelCase__		     =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      lowerCAmelCase__		     =     ['input_ids', 'attention_mask']






      def __init__(					self				:    Optional[int]						,						_A				:    Optional[Any]						,						_A				:    Optional[Any]="<s>"						,						_A				:    List[str]="</s>"						,						_A				:    Optional[Any]="</s>"						,						_A				:    List[str]="<s>"						,						_A				:    Tuple="<unk>"						,						_A				:    List[str]="<pad>"						,						_A				:    Optional[Dict[str, Any]] = None						,						**_A				:    Union[str, Any]						,						):


          '''simple docstring'''


          UpperCAmelCase__			:		str													=				{} if sp_model_kwargs is None else sp_model_kwargs

          # Compatibility with the original tokenizer
          UpperCAmelCase__			:		Optional[int]													=				7
          UpperCAmelCase__			:		Optional[int]													=				[f"""<madeupword{i}>""" for i in range(self.num_madeup_words )]

          UpperCAmelCase__			:		Tuple													=				kwargs.get('''additional_special_tokens'''						,						[] )
          kwargs["additional_special_tokens"] += [
              word for word in madeup_words if word not in kwargs["additional_special_tokens"]
          ]

          super().__init__(
              bos_token=_A						,						eos_token=_A						,						unk_token=_A						,						sep_token=_A						,						cls_token=_A						,						pad_token=_A						,						sp_model_kwargs=self.sp_model_kwargs						,						**_A						,						)

          UpperCAmelCase__			:		Any													=				spm.SentencePieceProcessor(**self.sp_model_kwargs )
          self.sp_model.Load(str(_A ) )
          UpperCAmelCase__			:		List[str]													=				vocab_file

          # Original fairseq vocab and spm vocab must be "aligned":
          # Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
          # -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
          # fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
          # spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

          # The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
          UpperCAmelCase__			:		Any													=				1

          # Mimic fairseq token-to-id alignment for the first 4 token
          UpperCAmelCase__			:		Any													=				{'''<s>''': 0, '''<pad>''': 1, '''</s>''': 2, '''<unk>''': 3}

          UpperCAmelCase__			:		int													=				len(self.sp_model )
          UpperCAmelCase__			:		Optional[int]													=				{f"""<madeupword{i}>""": sp_size + i + self.fairseq_offset for i in range(self.num_madeup_words )}
          self.fairseq_tokens_to_ids.update(_A )

          UpperCAmelCase__			:		Tuple													=				{v: k for k, v in self.fairseq_tokens_to_ids.items()}






      def __getstate__(					self				:    Any ):


          '''simple docstring'''


          UpperCAmelCase__			:		Dict													=				self.__dict__.copy()
          UpperCAmelCase__			:		Optional[int]													=				None
          UpperCAmelCase__			:		Optional[Any]													=				self.sp_model.serialized_model_proto()
          return state






      def __setstate__(					self				:    Union[str, Any]						,						_A				:    int ):


          '''simple docstring'''


          UpperCAmelCase__			:		List[Any]													=				d

          # for backward compatibility
          if not hasattr(self						,						'''sp_model_kwargs''' ):
              UpperCAmelCase__			:		Dict													=				{}

          UpperCAmelCase__			:		str													=				spm.SentencePieceProcessor(**self.sp_model_kwargs )
          self.sp_model.LoadFromSerializedProto(self.sp_model_proto )






      def lowercase_							(					self				:    Any						,						_A				:    List[int]						,						_A				:    Optional[List[int]] = None ):


          '''simple docstring'''


          if token_ids_a is None:
              return [self.sep_token_id] + token_ids_a
          UpperCAmelCase__			:		Dict													=				[self.sep_token_id]
          return sep + token_ids_a + sep + sep + token_ids_a






      def lowercase_							(					self				:    str						,						_A				:    List[int]						,						_A				:    Optional[List[int]] = None						,						_A				:    bool = False ):


          '''simple docstring'''


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

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






      def lowercase_							(					self				:    Any						,						_A				:    List[int]						,						_A				:    Optional[List[int]] = None ):


          '''simple docstring'''


          UpperCAmelCase__			:		Optional[int]													=				[self.sep_token_id]

          if token_ids_a is None:
              return len(sep + token_ids_a ) * [0]
          return len(sep + token_ids_a + sep + sep + token_ids_a ) * [0]






      @property
      def lowercase_							(					self				:    Any ):


          '''simple docstring'''


          return len(self.sp_model ) + self.fairseq_offset + self.num_madeup_words






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


          '''simple docstring'''


          UpperCAmelCase__			:		str													=				{self.convert_ids_to_tokens(_A ): i for i in range(self.vocab_size )}
          vocab.update(self.added_tokens_encoder )
          return vocab






      def lowercase_							(					self				:    Optional[Any]						,						_A				:    str ):


          '''simple docstring'''


          return self.sp_model.encode(_A						,						out_type=_A )






      def lowercase_							(					self				:    List[str]						,						_A				:    List[Any] ):


          '''simple docstring'''


          if token in self.fairseq_tokens_to_ids:
              return self.fairseq_tokens_to_ids[token]
          UpperCAmelCase__			:		Union[str, Any]													=				self.sp_model.PieceToId(_A )

          # Need to return unknown token if the SP model returned 0
          return spm_id + self.fairseq_offset if spm_id else self.unk_token_id






      def lowercase_							(					self				:    List[Any]						,						_A				:    str ):


          '''simple docstring'''


          if index in self.fairseq_ids_to_tokens:
              return self.fairseq_ids_to_tokens[index]
          return self.sp_model.IdToPiece(index - self.fairseq_offset )






      def lowercase_							(					self				:    int						,						_A				:    Optional[int] ):


          '''simple docstring'''


          UpperCAmelCase__			:		List[str]													=				''''''.join(_A ).replace(_A						,						''' ''' ).strip()
          return out_string






      def lowercase_							(					self				:    Any						,						_A				:    str						,						_A				:    Optional[str] = None ):


          '''simple docstring'''


          if not os.path.isdir(_A ):
              logger.error(f"""Vocabulary path ({save_directory}) should be a directory""" )
              return
          UpperCAmelCase__			:		List[str]													=				os.path.join(
              _A						,						(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file'''] )

          if os.path.abspath(self.vocab_file ) != os.path.abspath(_A ) and os.path.isfile(self.vocab_file ):
              copyfile(self.vocab_file						,						_A )
          elif not os.path.isfile(self.vocab_file ):
              with open(_A						,						'''wb''' ) as fi:
                  UpperCAmelCase__			:		List[Any]													=				self.sp_model.serialized_model_proto()
                  fi.write(_A )

          return (out_vocab_file,)
