




'''simple docstring'''







import os
import re
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


lowerCAmelCase_    	=							logging.get_logger(__name__)
lowerCAmelCase_    	=							{'''vocab_file''': '''spiece.model'''}

lowerCAmelCase_    	=							{
    '''vocab_file''': {
        '''google/fnet-base''': '''https://huggingface.co/google/fnet-base/resolve/main/spiece.model''',
        '''google/fnet-large''': '''https://huggingface.co/google/fnet-large/resolve/main/spiece.model''',
    },
}

lowerCAmelCase_    	=							{
    '''google/fnet-base''': 5_1_2,
    '''google/fnet-large''': 5_1_2,
}

lowerCAmelCase_    	=							'''▁'''

class   _snake_case( UpperCAmelCase			):
 __snake_case:       Tuple									=				VOCAB_FILES_NAMES
 __snake_case:       Optional[Any]									=				PRETRAINED_VOCAB_FILES_MAP
 __snake_case:       List[str]									=				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 __snake_case:       str									=				['''input_ids''', '''token_type_ids''']






 def __init__(self					:     Tuple						,		a					:     Optional[int]						,		a					:     Dict=False						,		a					:     Union[str, Any]=True						,		a					:     int=True						,		a					:     int="<unk>"						,		a					:     List[str]="[SEP]"						,		a					:     Union[str, Any]="<pad>"						,		a					:     Any="[CLS]"						,		a					:     Any="[MASK]"						,		a					:     Optional[Dict[str, Any]] = None						,		**a					:     List[str]						,		)					->      None:






    """simple docstring"""


    A__         =    (
        AddedToken(a						,		lstrip=a						,		rstrip=a						,		normalized=a						)
        if isinstance(a						,		a						)
        else mask_token
    )

    A__         =    {} if sp_model_kwargs is None else sp_model_kwargs

    super().__init__(
        do_lower_case=a						,		remove_space=a						,		keep_accents=a						,		unk_token=a						,		sep_token=a						,		pad_token=a						,		cls_token=a						,		mask_token=a						,		sp_model_kwargs=self.sp_model_kwargs						,		**a						,		)

    A__         =    do_lower_case
    A__         =    remove_space
    A__         =    keep_accents
    A__         =    vocab_file

    A__         =    spm.SentencePieceProcessor(**self.sp_model_kwargs						)
    self.sp_model.Load(a						)






 @property
 def 	_UpperCamelCase   (self					:     Dict						)					->      int:






    """simple docstring"""


    return len(self.sp_model						)






 def 	_UpperCamelCase   (self					:     Any						)					->      Optional[int]:






    """simple docstring"""


    A__         =    {self.convert_ids_to_tokens(a						): i for i in range(self.vocab_size						)}
    vocab.update(self.added_tokens_encoder						)
    return vocab






 def __getstate__(self					:     Any						)					->      Tuple:






    """simple docstring"""


    A__         =    self.__dict__.copy()
    A__         =    None
    return state






 def __setstate__(self					:     Tuple						,		a					:     int						)					->      Optional[int]:






    """simple docstring"""


    A__         =    d

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

    A__         =    spm.SentencePieceProcessor(**self.sp_model_kwargs						)
    self.sp_model.Load(self.vocab_file						)






 def 	_UpperCamelCase   (self					:     Dict						,		a					:     Optional[Any]						)					->      Optional[Any]:






    """simple docstring"""


    if self.remove_space:
       A__         =    ' '.join(inputs.strip().split()						)
    else:
       A__         =    inputs
    A__         =    outputs.replace('``'						,		'"'						).replace('\'\''						,		'"'						)

    if not self.keep_accents:
       A__         =    unicodedata.normalize('NFKD'						,		a						)
       A__         =    ''.join([c for c in outputs if not unicodedata.combining(a						)]						)
    if self.do_lower_case:
       A__         =    outputs.lower()

    return outputs






 def 	_UpperCamelCase   (self					:     Union[str, Any]						,		a					:     str						)					->      List[str]:






    """simple docstring"""


    A__         =    self.preprocess_text(a						)
    A__         =    self.sp_model.encode(a						,		out_type=a						)
    A__         =    []
    for piece in pieces:
       if len(a						) > 1 and piece[-1] == str(','						) and piece[-2].isdigit():
          A__         =    self.sp_model.EncodeAsPieces(piece[:-1].replace(a						,		''						)						)
          if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
             if len(cur_pieces[0]						) == 1:
                A__         =    cur_pieces[1:]
             else:
                A__         =    cur_pieces[0][1:]
          cur_pieces.append(piece[-1]						)
          new_pieces.extend(a						)
       else:
          new_pieces.append(a						)

    return new_pieces






 def 	_UpperCamelCase   (self					:     List[str]						,		a					:     Union[str, Any]						)					->      Optional[int]:






    """simple docstring"""


    return self.sp_model.PieceToId(a						)






 def 	_UpperCamelCase   (self					:     List[str]						,		a					:     str						)					->      Optional[int]:






    """simple docstring"""


    return self.sp_model.IdToPiece(a						)






 def 	_UpperCamelCase   (self					:     Optional[Any]						,		a					:     List[str]						)					->      Any:






    """simple docstring"""


    A__         =    []
    A__         =    ''
    A__         =    False
    for token in tokens:
       # make sure that special tokens are not decoded using sentencepiece model
       if token in self.all_special_tokens:
          if not prev_is_special:
             out_string += " "
          out_string += self.sp_model.decode(a						) + token
          A__         =    True
          A__         =    []
       else:
          current_sub_tokens.append(a						)
          A__         =    False
    out_string += self.sp_model.decode(a						)
    return out_string.strip()






 def 	_UpperCamelCase   (self					:     Optional[int]						,		a					:     List[int]						,		a					:     bool = False						,		a					:     bool = None						,		a					:     bool = True						,		**a					:     List[Any]						,		)					->      str:






    """simple docstring"""


    A__         =    kwargs.pop('use_source_tokenizer'						,		a						)

    A__         =    self.convert_ids_to_tokens(a						,		skip_special_tokens=a						)

    # To avoid mixing byte-level and unicode for byte-level BPT
    # we need to build string separately for added tokens and byte-level tokens
    # cf. https://github.com/huggingface/transformers/issues/1133
    A__         =    []
    A__         =    []
    for token in filtered_tokens:
       if skip_special_tokens and token in self.all_special_ids:
          continue
       if token in self.added_tokens_encoder:
          if current_sub_text:
             sub_texts.append(self.convert_tokens_to_string(a						)						)
             A__         =    []
          sub_texts.append(a						)
       else:
          current_sub_text.append(a						)
    if current_sub_text:
       sub_texts.append(self.convert_tokens_to_string(a						)						)

    # Mimic the behavior of the Rust tokenizer:
    # No space after <unk>
    if spaces_between_special_tokens:
       A__         =    re.sub(R'(<unk>) '						,		R'\1'						,		' '.join(a						)						)
    else:
       A__         =    ''.join(a						)

    A__         =    (
        clean_up_tokenization_spaces
        if clean_up_tokenization_spaces is not None
        else self.clean_up_tokenization_spaces
    )
    if clean_up_tokenization_spaces:
       A__         =    self.clean_up_tokenization(a						)
       return clean_text
    else:
       return text






 def 	_UpperCamelCase   (self					:     int						,		a					:     List[int]						,		a					:     Optional[List[int]] = None						)					->      List[int]:






    """simple docstring"""


    A__         =    [self.sep_token_id]
    A__         =    [self.cls_token_id]
    if token_ids_a is None:
       return cls + token_ids_a + sep
    return cls + token_ids_a + sep + token_ids_a + sep






 def 	_UpperCamelCase   (self					:     Tuple						,		a					:     List[int]						,		a					:     Optional[List[int]] = None						,		a					:     bool = False						)					->      List[int]:






    """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 not None:
       return [1] + ([0] * len(a						)) + [1] + ([0] * len(a						)) + [1]
    return [1] + ([0] * len(a						)) + [1]






 def 	_UpperCamelCase   (self					:     Union[str, Any]						,		a					:     List[int]						,		a					:     Optional[List[int]] = None						)					->      List[int]:






    """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						) * [0] + len(token_ids_a + sep						) * [1]






 def 	_UpperCamelCase   (self					:     List[str]						,		a					:     str						,		a					:     Optional[str] = None						)					->      Tuple[str]:






    """simple docstring"""


    if not os.path.isdir(a						):
       logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""						)
       return
    A__         =    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:
          A__         =    self.sp_model.serialized_model_proto()
          fi.write(a						)

    return (out_vocab_file,)








'''simple docstring'''







import itertools
import random
import unittest

import numpy as np

from transformers import EncodecFeatureExtractor
from transformers.testing_utils import require_torch
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
    import torch


lowerCAmelCase_    	=							random.Random()







def 	_A			(   UpperCAmelCase			,UpperCAmelCase=1.0			,UpperCAmelCase=None			,UpperCAmelCase=None	):






   '''simple docstring'''







   if rng is None:
      A__         =    global_rng

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

   return values

@require_torch
class   _snake_case( unittest.TestCase			):






 def __init__(self					:     int						,		a					:     str						,		a					:     Optional[int]=7						,		a					:     Optional[Any]=4_00						,		a					:     Any=20_00						,		a					:     Optional[Any]=1						,		a					:     Optional[int]=0.0						,		a					:     Dict=2_40_00						,		a					:     Tuple=True						,		)					->      Optional[Any]:






    """simple docstring"""


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






 def 	_UpperCamelCase   (self					:     int						)					->      List[str]:






    """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,
    }






 def 	_UpperCamelCase   (self					:     List[Any]						,		a					:     Any=False						,		a					:     Optional[Any]=False						)					->      Tuple:






    """simple docstring"""


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

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

    return audio_inputs




@require_torch
class   _snake_case( UpperCAmelCase					,   unittest.TestCase			):
 __snake_case:       Union[str, Any]									=				EncodecFeatureExtractor






 def 	_UpperCamelCase   (self					:     Union[str, Any]						)					->      Optional[Any]:






    """simple docstring"""


    A__         =    EnCodecFeatureExtractionTester(self						)






 def 	_UpperCamelCase   (self					:     Dict						)					->      Optional[Any]:






    """simple docstring"""


    A__         =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
    # create three inputs of length 800, 1000, and 1200
    A__         =    [floats_list((1, x)						)[0] for x in range(8_00						,		14_00						,		2_00						)]
    A__         =    [np.asarray(a						) for audio_input in audio_inputs]

    # Test not batched input
    A__         =    feat_extract(audio_inputs[0]						,		return_tensors='np'						).input_values
    A__         =    feat_extract(np_audio_inputs[0]						,		return_tensors='np'						).input_values
    self.assertTrue(np.allclose(a						,		a						,		atol=1e-3						)						)

    # Test batched
    A__         =    feat_extract(a						,		padding=a						,		return_tensors='np'						).input_values
    A__         =    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						)						)






 def 	_UpperCamelCase   (self					:     Tuple						)					->      Any:






    """simple docstring"""


    A__         =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
    A__         =    np.random.rand(1_00						).astype(np.floataa						)
    A__         =    np_audio_inputs.tolist()

    for inputs in [py_audio_inputs, np_audio_inputs]:
       A__         =    feature_extractor.pad([{'input_values': inputs}]						,		return_tensors='np'						)
       self.assertTrue(np_processed.input_values.dtype == np.floataa						)
       A__         =    feature_extractor.pad([{'input_values': inputs}]						,		return_tensors='pt'						)
       self.assertTrue(pt_processed.input_values.dtype == torch.floataa						)






 def 	_UpperCamelCase   (self					:     Optional[int]						,		a					:     Optional[Any]						)					->      Union[str, Any]:






    """simple docstring"""


    from datasets import load_dataset

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

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






 def 	_UpperCamelCase   (self					:     Tuple						)					->      Tuple:






    """simple docstring"""


    A__         =    torch.tensor(
        [2.3804e-03, 2.0752e-03, 1.9836e-03, 2.1057e-03, 1.6174e-03,
         3.0518e-04, 9.1553e-05, 3.3569e-04, 9.7656e-04, 1.8311e-03,
         2.0142e-03, 2.1057e-03, 1.7395e-03, 4.5776e-04, -3.9673e-04,
         4.5776e-04, 1.0071e-03, 9.1553e-05, 4.8828e-04, 1.1597e-03,
         7.3242e-04, 9.4604e-04, 1.8005e-03, 1.8311e-03, 8.8501e-04,
         4.2725e-04, 4.8828e-04, 7.3242e-04, 1.0986e-03, 2.1057e-03]						)
    # fmt: on
    A__         =    self._load_datasamples(1						)
    A__         =    EncodecFeatureExtractor()
    A__         =    feature_extractor(a						,		return_tensors='pt'						).input_values
    self.assertEquals(input_values.shape						,		(1, 1, 9_36_80)						)
    self.assertTrue(torch.allclose(input_values[0, 0, :30]						,		a						,		atol=1e-6						)						)






 def 	_UpperCamelCase   (self					:     int						)					->      Optional[Any]:






    """simple docstring"""


    A__         =    torch.tensor(
        [2.3804e-03, 2.0752e-03, 1.9836e-03, 2.1057e-03, 1.6174e-03,
         3.0518e-04, 9.1553e-05, 3.3569e-04, 9.7656e-04, 1.8311e-03,
         2.0142e-03, 2.1057e-03, 1.7395e-03, 4.5776e-04, -3.9673e-04,
         4.5776e-04, 1.0071e-03, 9.1553e-05, 4.8828e-04, 1.1597e-03,
         7.3242e-04, 9.4604e-04, 1.8005e-03, 1.8311e-03, 8.8501e-04,
         4.2725e-04, 4.8828e-04, 7.3242e-04, 1.0986e-03, 2.1057e-03]						)
    # fmt: on
    A__         =    self._load_datasamples(1						)
    A__         =    [np.tile(input_audio[0][None]						,		reps=(2, 1)						)]
    input_audio[0][1] *= 0.5
    A__         =    EncodecFeatureExtractor(feature_size=2						)
    A__         =    feature_extractor(a						,		return_tensors='pt'						).input_values
    self.assertEquals(input_values.shape						,		(1, 2, 9_36_80)						)
    self.assertTrue(torch.allclose(input_values[0, 0, :30]						,		a						,		atol=1e-6						)						)
    self.assertTrue(torch.allclose(input_values[0, 1, :30]						,		EXPECTED_INPUT_VALUES * 0.5						,		atol=1e-6						)						)






 def 	_UpperCamelCase   (self					:     Union[str, Any]						)					->      Any:






    """simple docstring"""


    A__         =    self._load_datasamples(2						)
    # would be easier if the stride was like
    A__         =    EncodecFeatureExtractor(feature_size=1						,		chunk_length_s=1						,		overlap=0.01						)

    # pad and trunc raise an error ?
    with self.assertRaisesRegex(
        a						,		'^Both padding and truncation were set. Make sure you only set one.$'						,		):
       A__         =    feature_extractor(
           a						,		padding='max_length'						,		truncation=a						,		return_tensors='pt'						).input_values

    # truncate to chunk
    A__         =    feature_extractor(a						,		truncation=a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(2, 1, 7_15_20)						)  # 2 chunks

    # force truncate to max_length
    A__         =    feature_extractor(
        a						,		truncation=a						,		max_length=4_80_00						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(2, 1, 4_80_00)						)

    # pad to chunk
    A__         =    feature_extractor(a						,		padding=a						,		return_tensors='pt'						).input_values
    self.assertEquals(padded_outputs.shape						,		(2, 1, 9_52_80)						)

    # pad to chunk
    A__         =    feature_extractor(a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(2, 1, 9_52_80)						)

    # force pad to max length
    A__         =    feature_extractor(
        a						,		padding='max_length'						,		max_length=10_00_00						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(2, 1, 10_00_00)						)

    # force no pad
    with self.assertRaisesRegex(
        a						,		'^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'						,		):
       A__         =    feature_extractor(a						,		padding=a						,		return_tensors='pt'						).input_values

    A__         =    feature_extractor(input_audio[0]						,		padding=a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(1, 1, 9_36_80)						)

    # no pad if no chunk_length_s
    A__         =    None
    with self.assertRaisesRegex(
        a						,		'^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'						,		):
       A__         =    feature_extractor(a						,		padding=a						,		return_tensors='pt'						).input_values

    A__         =    feature_extractor(input_audio[0]						,		padding=a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(1, 1, 9_36_80)						)

    # no pad if no overlap
    A__         =    2
    A__         =    None
    with self.assertRaisesRegex(
        a						,		'^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'						,		):
       A__         =    feature_extractor(a						,		padding=a						,		return_tensors='pt'						).input_values

    A__         =    feature_extractor(input_audio[0]						,		padding=a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(1, 1, 9_36_80)						)


