




import os
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 SPIECE_UNDERLINE, logging


_UpperCAmelCase	:						List[str]			=     logging.get_logger(__name__)

_UpperCAmelCase	:						Tuple			=     {"vocab_file": "spiece.model"}

_UpperCAmelCase	:						str			=     {
    "vocab_file": {
        "xlnet-base-cased": "https://huggingface.co/xlnet-base-cased/resolve/main/spiece.model",
        "xlnet-large-cased": "https://huggingface.co/xlnet-large-cased/resolve/main/spiece.model",
    }
}

_UpperCAmelCase	:						Union[str, Any]			=     {
    "xlnet-base-cased": None,
    "xlnet-large-cased": None,
}

# Segments (not really needed)
_UpperCAmelCase	:						int			=     0
_UpperCAmelCase	:						int			=     1
_UpperCAmelCase	:						List[Any]			=     2
_UpperCAmelCase	:						Union[str, Any]			=     3
_UpperCAmelCase	:						Optional[Any]			=     4




class    __lowerCAmelCase       (							lowerCAmelCase):
       _a 					=						VOCAB_FILES_NAMES
       _a 					=						PRETRAINED_VOCAB_FILES_MAP
       _a 					=						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       _a 					=						'''left'''






       def __init__(   self:					Optional[Any]       ,					_lowerCAmelCase:					List[str]       ,					_lowerCAmelCase:					int=False       ,					_lowerCAmelCase:					Tuple=True       ,					_lowerCAmelCase:					Any=False       ,					_lowerCAmelCase:					str="<s>"       ,					_lowerCAmelCase:					Union[str, Any]="</s>"       ,					_lowerCAmelCase:					List[Any]="<unk>"       ,					_lowerCAmelCase:					int="<sep>"       ,					_lowerCAmelCase:					Optional[int]="<pad>"       ,					_lowerCAmelCase:					Tuple="<cls>"       ,					_lowerCAmelCase:					str="<mask>"       ,					_lowerCAmelCase:					List[Any]=["<eop>", "<eod>"]       ,					_lowerCAmelCase:					Optional[Dict[str, Any]] = None       ,					**_lowerCAmelCase:					Optional[Any]       ,					):
         # Mask token behave like a normal word, i.e. include the space before it
         lowercase    :Dict      	=						AddedToken(_lowerCAmelCase       ,					lstrip=_lowerCAmelCase       ,					rstrip=_lowerCAmelCase				) if isinstance(_lowerCAmelCase       ,					_lowerCAmelCase				) else mask_token

         lowercase    :int      	=						{} if sp_model_kwargs is None else sp_model_kwargs

         super().__init__(
             do_lower_case=_lowerCAmelCase       ,					remove_space=_lowerCAmelCase       ,					keep_accents=_lowerCAmelCase       ,					bos_token=_lowerCAmelCase       ,					eos_token=_lowerCAmelCase       ,					unk_token=_lowerCAmelCase       ,					sep_token=_lowerCAmelCase       ,					pad_token=_lowerCAmelCase       ,					cls_token=_lowerCAmelCase       ,					mask_token=_lowerCAmelCase       ,					additional_special_tokens=_lowerCAmelCase       ,					sp_model_kwargs=self.sp_model_kwargs       ,					**_lowerCAmelCase       ,					)

         lowercase    :Dict      	=						3

         lowercase    :Tuple      	=						do_lower_case
         lowercase    :str      	=						remove_space
         lowercase    :str      	=						keep_accents
         lowercase    :Optional[int]      	=						vocab_file

         lowercase    :List[str]      	=						spm.SentencePieceProcessor(**self.sp_model_kwargs				)
         self.sp_model.Load(_lowerCAmelCase				)






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Any				):
         return len(self.sp_model				)






       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]				):
         lowercase    :Optional[int]      	=						{self.convert_ids_to_tokens(_lowerCAmelCase				): i for i in range(self.vocab_size				)}
         vocab.update(self.added_tokens_encoder				)
         return vocab






       def __getstate__(   self:					Dict				):
         lowercase    :Any      	=						self.__dict__.copy()
         lowercase    :Optional[int]      	=						None
         return state






       def __setstate__(   self:					Union[str, Any]       ,					_lowerCAmelCase:					Any				):
         lowercase    :List[Any]      	=						d

         # for backward compatibility
         if not hasattr(self       ,					"sp_model_kwargs"				):
           lowercase    :int      	=						{}

         lowercase    :Any      	=						spm.SentencePieceProcessor(**self.sp_model_kwargs				)
         self.sp_model.Load(self.vocab_file				)






       def 					SCREAMING_SNAKE_CASE    (   self:					Dict       ,					_lowerCAmelCase:					List[Any]				):
         if self.remove_space:
           lowercase    :Any      	=						" ".join(inputs.strip().split()				)
         else:
           lowercase    :Any      	=						inputs
         lowercase    :Dict      	=						outputs.replace("``"       ,					"\""				).replace("''"       ,					"\""				)

         if not self.keep_accents:
           lowercase    :List[Any]      	=						unicodedata.normalize("NFKD"       ,					_lowerCAmelCase				)
           lowercase    :Optional[int]      	=						"".join([c for c in outputs if not unicodedata.combining(_lowerCAmelCase				)]				)
         if self.do_lower_case:
           lowercase    :str      	=						outputs.lower()

         return outputs






       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[Any]       ,					_lowerCAmelCase:					str				):
         lowercase    :Optional[Any]      	=						self.preprocess_text(_lowerCAmelCase				)
         lowercase    :List[Any]      	=						self.sp_model.encode(_lowerCAmelCase       ,					out_type=_lowerCAmelCase				)
         lowercase    :str      	=						[]
         for piece in pieces:
           if len(_lowerCAmelCase				) > 1 and piece[-1] == str(","				) and piece[-2].isdigit():
             lowercase    :List[Any]      	=						self.sp_model.EncodeAsPieces(piece[:-1].replace(_lowerCAmelCase       ,					""				)				)
             if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
               if len(cur_pieces[0]				) == 1:
                 lowercase    :Tuple      	=						cur_pieces[1:]
               else:
                 lowercase    :List[str]      	=						cur_pieces[0][1:]
             cur_pieces.append(piece[-1]				)
             new_pieces.extend(_lowerCAmelCase				)
           else:
             new_pieces.append(_lowerCAmelCase				)

         return new_pieces






       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]       ,					_lowerCAmelCase:					str				):
         return self.sp_model.PieceToId(_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					str       ,					_lowerCAmelCase:					Any				):
         return self.sp_model.IdToPiece(_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[int]       ,					_lowerCAmelCase:					Union[str, Any]				):
         lowercase    :List[Any]      	=						"".join(_lowerCAmelCase				).replace(_lowerCAmelCase       ,					" "				).strip()
         return out_string






       def 					SCREAMING_SNAKE_CASE    (   self:					Union[str, Any]       ,					_lowerCAmelCase:					List[int]       ,					_lowerCAmelCase:					bool = False       ,					_lowerCAmelCase:					bool = None       ,					_lowerCAmelCase:					bool = True       ,					**_lowerCAmelCase:					str       ,					):
         lowercase    :Optional[Any]      	=						kwargs.pop("use_source_tokenizer"       ,					_lowerCAmelCase				)

         lowercase    :str      	=						self.convert_ids_to_tokens(_lowerCAmelCase       ,					skip_special_tokens=_lowerCAmelCase				)

         # 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
         lowercase    :str      	=						[]
         lowercase    :int      	=						[]
         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(_lowerCAmelCase				)				)
               lowercase    :Union[str, Any]      	=						[]
             sub_texts.append(_lowerCAmelCase				)
           else:
             current_sub_text.append(_lowerCAmelCase				)
         if current_sub_text:
           sub_texts.append(self.convert_tokens_to_string(_lowerCAmelCase				)				)

         # Mimic the behavior of the Rust tokenizer:
         # By default, there are no spaces between special tokens
         lowercase    :Dict      	=						"".join(_lowerCAmelCase				)

         lowercase    :Dict      	=						(
             clean_up_tokenization_spaces
             if clean_up_tokenization_spaces is not None
             else self.clean_up_tokenization_spaces
         )
         if clean_up_tokenization_spaces:
           lowercase    :int      	=						self.clean_up_tokenization(_lowerCAmelCase				)
           return clean_text
         else:
           return text






       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[Any]       ,					_lowerCAmelCase:					List[int]       ,					_lowerCAmelCase:					Optional[List[int]] = None				):
         lowercase    :List[Any]      	=						[self.sep_token_id]
         lowercase    :List[Any]      	=						[self.cls_token_id]
         if token_ids_a is None:
           return token_ids_a + sep + cls
         return token_ids_a + sep + token_ids_a + sep + cls






       def 					SCREAMING_SNAKE_CASE    (   self:					Union[str, Any]       ,					_lowerCAmelCase:					List[int]       ,					_lowerCAmelCase:					Optional[List[int]] = None       ,					_lowerCAmelCase:					bool = False				):

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






       def 					SCREAMING_SNAKE_CASE    (   self:					Tuple       ,					_lowerCAmelCase:					List[int]       ,					_lowerCAmelCase:					Optional[List[int]] = None				):
         lowercase    :List[str]      	=						[self.sep_token_id]
         lowercase    :Dict      	=						[2]

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






       def 					SCREAMING_SNAKE_CASE    (   self:					int       ,					_lowerCAmelCase:					str       ,					_lowerCAmelCase:					Optional[str] = None				):
         if not os.path.isdir(_lowerCAmelCase				):
           logger.error(F"Vocabulary path ({save_directory}) should be a directory"				)
           return
         lowercase    :Optional[int]      	=						os.path.join(
             _lowerCAmelCase       ,					(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]				)

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

         return (out_vocab_file,)









import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
     import torch

if is_datasets_available():
     from datasets import load_dataset

_UpperCAmelCase	:						List[str]			=     random.Random()



def       UpperCAmelCase__  (     lowerCamelCase,			lowerCamelCase=1.0,			lowerCamelCase=None,			lowerCamelCase=None   ):
  if rng is None:
    lowercase    :Union[str, Any]      	=						global_rng

  lowercase    :Any      	=						[]
  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:					Optional[Any]       ,					_lowerCAmelCase:					Tuple       ,					_lowerCAmelCase:					Union[str, Any]=7       ,					_lowerCAmelCase:					str=4_00       ,					_lowerCAmelCase:					List[str]=20_00       ,					_lowerCAmelCase:					Dict=20_48       ,					_lowerCAmelCase:					Any=1_28       ,					_lowerCAmelCase:					Any=1       ,					_lowerCAmelCase:					List[Any]=5_12       ,					_lowerCAmelCase:					Optional[int]=30       ,					_lowerCAmelCase:					List[Any]=4_41_00       ,					):
         lowercase    :Any      	=						parent
         lowercase    :Any      	=						batch_size
         lowercase    :Optional[int]      	=						min_seq_length
         lowercase    :Optional[Any]      	=						max_seq_length
         lowercase    :Dict      	=						(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
         lowercase    :List[str]      	=						spectrogram_length
         lowercase    :int      	=						feature_size
         lowercase    :Union[str, Any]      	=						num_audio_channels
         lowercase    :Optional[int]      	=						hop_length
         lowercase    :str      	=						chunk_length
         lowercase    :List[str]      	=						sampling_rate






       def 					SCREAMING_SNAKE_CASE    (   self:					str				):
         return {
             "spectrogram_length": self.spectrogram_length,
             "feature_size": self.feature_size,
             "num_audio_channels": self.num_audio_channels,
             "hop_length": self.hop_length,
             "chunk_length": self.chunk_length,
             "sampling_rate": self.sampling_rate,
         }






       def 					SCREAMING_SNAKE_CASE    (   self:					int       ,					_lowerCAmelCase:					int=False       ,					_lowerCAmelCase:					int=False				):
         def _flatten(_lowerCAmelCase:					Optional[Any]				):
           return list(itertools.chain(*_lowerCAmelCase				)				)

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







@require_torch
@require_torchaudio
class    __lowerCAmelCase       (							lowerCAmelCase      , unittest.TestCase):
       _a 					=						TvltFeatureExtractor






       def 					SCREAMING_SNAKE_CASE    (   self:					Tuple				):
         lowercase    :Dict      	=						TvltFeatureExtractionTester(self				)






       def 					SCREAMING_SNAKE_CASE    (   self:					str				):
         lowercase    :List[Any]      	=						self.feature_extraction_class(**self.feat_extract_dict				)
         self.assertTrue(hasattr(_lowerCAmelCase       ,					"spectrogram_length"				)				)
         self.assertTrue(hasattr(_lowerCAmelCase       ,					"feature_size"				)				)
         self.assertTrue(hasattr(_lowerCAmelCase       ,					"num_audio_channels"				)				)
         self.assertTrue(hasattr(_lowerCAmelCase       ,					"hop_length"				)				)
         self.assertTrue(hasattr(_lowerCAmelCase       ,					"chunk_length"				)				)
         self.assertTrue(hasattr(_lowerCAmelCase       ,					"sampling_rate"				)				)






       def 					SCREAMING_SNAKE_CASE    (   self:					str				):
         lowercase    :Any      	=						self.feature_extraction_class(**self.feat_extract_dict				)

         with tempfile.TemporaryDirectory() as tmpdirname:
           lowercase    :Tuple      	=						feat_extract_first.save_pretrained(_lowerCAmelCase				)[0]
           check_json_file_has_correct_format(_lowerCAmelCase				)
           lowercase    :Tuple      	=						self.feature_extraction_class.from_pretrained(_lowerCAmelCase				)

         lowercase    :Optional[Any]      	=						feat_extract_first.to_dict()
         lowercase    :List[str]      	=						feat_extract_second.to_dict()
         lowercase    :Any      	=						dict_first.pop("mel_filters"				)
         lowercase    :Any      	=						dict_second.pop("mel_filters"				)
         self.assertTrue(np.allclose(_lowerCAmelCase       ,					_lowerCAmelCase				)				)
         self.assertEqual(_lowerCAmelCase       ,					_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					Tuple				):
         lowercase    :Union[str, Any]      	=						self.feature_extraction_class(**self.feat_extract_dict				)

         with tempfile.TemporaryDirectory() as tmpdirname:
           lowercase    :Any      	=						os.path.join(_lowerCAmelCase       ,					"feat_extract.json"				)
           feat_extract_first.to_json_file(_lowerCAmelCase				)
           lowercase    :List[Any]      	=						self.feature_extraction_class.from_json_file(_lowerCAmelCase				)

         lowercase    :Tuple      	=						feat_extract_first.to_dict()
         lowercase    :int      	=						feat_extract_second.to_dict()
         lowercase    :Tuple      	=						dict_first.pop("mel_filters"				)
         lowercase    :Tuple      	=						dict_second.pop("mel_filters"				)
         self.assertTrue(np.allclose(_lowerCAmelCase       ,					_lowerCAmelCase				)				)
         self.assertEqual(_lowerCAmelCase       ,					_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					int				):
         # Initialize feature_extractor
         lowercase    :int      	=						self.feature_extraction_class(**self.feat_extract_dict				)

         # create three inputs of length 800, 1000, and 1200
         lowercase    :List[Any]      	=						[floats_list((1, x)				)[0] for x in range(8_00       ,					14_00       ,					2_00				)]
         lowercase    :Union[str, Any]      	=						[np.asarray(_lowerCAmelCase				) for speech_input in speech_inputs]

         # Test not batched input
         lowercase    :Any      	=						feature_extractor(np_speech_inputs[0]       ,					return_tensors="np"       ,					sampling_rate=4_41_00				).audio_values

         self.assertTrue(encoded_audios.ndim == 4				)
         self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
         self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
         self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)

         # Test batched
         lowercase    :Optional[int]      	=						feature_extractor(_lowerCAmelCase       ,					return_tensors="np"       ,					sampling_rate=4_41_00				).audio_values

         self.assertTrue(encoded_audios.ndim == 4				)
         self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
         self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
         self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)

         # Test audio masking
         lowercase    :Dict      	=						feature_extractor(
             _lowerCAmelCase       ,					return_tensors="np"       ,					sampling_rate=4_41_00       ,					mask_audio=_lowerCAmelCase				).audio_values

         self.assertTrue(encoded_audios.ndim == 4				)
         self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
         self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
         self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)

         # Test 2-D numpy arrays are batched.
         lowercase    :Optional[int]      	=						[floats_list((1, x)				)[0] for x in (8_00, 8_00, 8_00)]
         lowercase    :Union[str, Any]      	=						np.asarray(_lowerCAmelCase				)
         lowercase    :Dict      	=						feature_extractor(_lowerCAmelCase       ,					return_tensors="np"       ,					sampling_rate=4_41_00				).audio_values
         self.assertTrue(encoded_audios.ndim == 4				)
         self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
         self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
         self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)






       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[int]       ,					_lowerCAmelCase:					Optional[Any]				):
         lowercase    :str      	=						load_dataset("hf-internal-testing/librispeech_asr_dummy"       ,					"clean"       ,					split="validation"				)
         # automatic decoding with librispeech
         lowercase    :Any      	=						ds.sort("id"				).select(range(_lowerCAmelCase				)				)[:num_samples]["audio"]

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






       def 					SCREAMING_SNAKE_CASE    (   self:					Dict				):
         lowercase    :Optional[int]      	=						self._load_datasamples(1				)
         lowercase    :Union[str, Any]      	=						TvltFeatureExtractor()
         lowercase    :str      	=						feature_extractor(_lowerCAmelCase       ,					return_tensors="pt"				).audio_values

         self.assertEquals(audio_values.shape       ,					(1, 1, 1_92, 1_28)				)

         lowercase    :Optional[int]      	=						torch.tensor([[-0.30_32, -0.27_08], [-0.44_34, -0.40_07]]				)
         self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]       ,					_lowerCAmelCase       ,					atol=1e-4				)				)



