

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


__a    :List[Any]      =			logging.get_logger(__name__)

__a    :List[Any]      =			{'vocab_file': 'spiece.model'}

__a    :Optional[Any]      =			{
    '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',
    }
}

__a    :int      =			{
    'xlnet-base-cased': None,
    'xlnet-large-cased': None,
}

# Segments (not really needed)
__a    :Optional[Any]      =			0
__a    :Optional[Any]      =			1
__a    :List[Any]      =			2
__a    :Dict      =			3
__a    :List[Any]      =			4

class   _a					(			snake_case_				):




      """simple docstring"""

      _lowerCamelCase       :			List[str]      		=       VOCAB_FILES_NAMES
      _lowerCamelCase       :			Optional[Any]      		=       PRETRAINED_VOCAB_FILES_MAP
      _lowerCamelCase       :			int      		=       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _lowerCamelCase       :			str      		=       'left'





      def __init__(   self							:				Tuple						,			UpperCAmelCase							:				str						,			UpperCAmelCase							:				Tuple=False						,			UpperCAmelCase							:				Tuple=True						,			UpperCAmelCase							:				Tuple=False						,			UpperCAmelCase							:				Dict="<s>"						,			UpperCAmelCase							:				Dict="</s>"						,			UpperCAmelCase							:				Optional[int]="<unk>"						,			UpperCAmelCase							:				Tuple="<sep>"						,			UpperCAmelCase							:				Any="<pad>"						,			UpperCAmelCase							:				Union[str, Any]="<cls>"						,			UpperCAmelCase							:				Optional[int]="<mask>"						,			UpperCAmelCase							:				List[Any]=["<eop>", "<eod>"]						,			UpperCAmelCase							:				Dict = None						,			**UpperCAmelCase							:				Tuple						,			):
            A_						  =				AddedToken(UpperCAmelCase						,			lstrip=UpperCAmelCase						,			rstrip=UpperCAmelCase	) if isinstance(UpperCAmelCase						,			UpperCAmelCase	) else mask_token

            A_						  =				{} if sp_model_kwargs is None else sp_model_kwargs

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

            A_						  =				3

            A_						  =				do_lower_case
            A_						  =				remove_space
            A_						  =				keep_accents
            A_						  =				vocab_file

            A_						  =				spm.SentencePieceProcessor(**self.sp_model_kwargs	)
            self.sp_model.Load(UpperCAmelCase	)





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





      def 		__A	(   self							:				Any	):
            A_						  =				{self.convert_ids_to_tokens(UpperCAmelCase	): i for i in range(self.vocab_size	)}
            vocab.update(self.added_tokens_encoder	)
            return vocab





      def __getstate__(   self							:				Any	):
            A_						  =				self.__dict__.copy()
            A_						  =				None
            return state





      def __setstate__(   self							:				str						,			UpperCAmelCase							:				Dict	):
            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 		__A	(   self							:				List[str]						,			UpperCAmelCase							:				Optional[Any]	):

            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"						,			UpperCAmelCase	)
                  A_						  =				"".join([c for c in outputs if not unicodedata.combining(UpperCAmelCase	)]	)
            if self.do_lower_case:
                  A_						  =				outputs.lower()

            return outputs





      def 		__A	(   self							:				List[str]						,			UpperCAmelCase							:				str	):
            A_						  =				self.preprocess_text(UpperCAmelCase	)
            A_						  =				self.sp_model.encode(UpperCAmelCase						,			out_type=UpperCAmelCase	)
            A_						  =				[]
            for piece in pieces:
                  if len(UpperCAmelCase	) > 1 and piece[-1] == str(","	) and piece[-2].isdigit():
                        A_						  =				self.sp_model.EncodeAsPieces(piece[:-1].replace(UpperCAmelCase						,			""	)	)
                        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(UpperCAmelCase	)
                  else:
                        new_pieces.append(UpperCAmelCase	)

            return new_pieces





      def 		__A	(   self							:				List[Any]						,			UpperCAmelCase							:				Tuple	):
            return self.sp_model.PieceToId(UpperCAmelCase	)





      def 		__A	(   self							:				Tuple						,			UpperCAmelCase							:				Union[str, Any]	):
            return self.sp_model.IdToPiece(UpperCAmelCase	)





      def 		__A	(   self							:				str						,			UpperCAmelCase							:				int	):
            A_						  =				"".join(UpperCAmelCase	).replace(UpperCAmelCase						,			" "	).strip()
            return out_string





      def 		__A	(   self							:				List[Any]						,			UpperCAmelCase							:				Union[str, Any]						,			UpperCAmelCase							:				Optional[int] = False						,			UpperCAmelCase							:				str = None						,			UpperCAmelCase							:				Dict = True						,			**UpperCAmelCase							:				Optional[Any]						,			):
            A_						  =				kwargs.pop("use_source_tokenizer"						,			UpperCAmelCase	)

            A_						  =				self.convert_ids_to_tokens(UpperCAmelCase						,			skip_special_tokens=UpperCAmelCase	)

            # 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(UpperCAmelCase	)	)
                              A_						  =				[]
                        sub_texts.append(UpperCAmelCase	)
                  else:
                        current_sub_text.append(UpperCAmelCase	)
            if current_sub_text:
                  sub_texts.append(self.convert_tokens_to_string(UpperCAmelCase	)	)

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

            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(UpperCAmelCase	)
                  return clean_text
            else:
                  return text





      def 		__A	(   self							:				int						,			UpperCAmelCase							:				Tuple						,			UpperCAmelCase							:				str = None	):
            A_						  =				[self.sep_token_id]
            A_						  =				[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 		__A	(   self							:				Dict						,			UpperCAmelCase							:				int						,			UpperCAmelCase							:				Tuple = None						,			UpperCAmelCase							:				Optional[int] = False	):

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

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





      def 		__A	(   self							:				Union[str, Any]						,			UpperCAmelCase							:				Union[str, Any]						,			UpperCAmelCase							:				List[Any] = None	):
            A_						  =				[self.sep_token_id]
            A_						  =				[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 		__A	(   self							:				Optional[int]						,			UpperCAmelCase							:				Any						,			UpperCAmelCase							:				Optional[int] = None	):

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

            if os.path.abspath(self.vocab_file	) != os.path.abspath(UpperCAmelCase	) and os.path.isfile(self.vocab_file	):
                  copyfile(self.vocab_file						,			UpperCAmelCase	)
            elif not os.path.isfile(self.vocab_file	):
                  with open(UpperCAmelCase						,			"wb"	) as fi:
                        A_						  =				self.sp_model.serialized_model_proto()
                        fi.write(UpperCAmelCase	)

            return (out_vocab_file,)


"""simple docstring"""

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		     =    random.Random()



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__=1.0					,					lowercase__=None					,					lowercase__=None   )			->						Optional[Any]:
 if rng is None:
  lowerCAmelCase__   :				Union[str, Any] 			=				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 			__a					(						unittest.TestCase       ):





       """simple docstring"""


       def __init__(   self			,					snake_case			,					snake_case=7			,					snake_case=400			,					snake_case=2_000			,					snake_case=2_048			,					snake_case=128			,					snake_case=1			,					snake_case=512			,					snake_case=30			,					snake_case=44_100			,					):





        """simple docstring"""

        lowerCAmelCase__   :				Optional[int] 			=				parent
        lowerCAmelCase__   :				Optional[int] 			=				batch_size
        lowerCAmelCase__   :				Optional[Any] 			=				min_seq_length
        lowerCAmelCase__   :				Optional[int] 			=				max_seq_length
        lowerCAmelCase__   :				Dict 			=				(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        lowerCAmelCase__   :				Dict 			=				spectrogram_length
        lowerCAmelCase__   :				Any 			=				feature_size
        lowerCAmelCase__   :				int 			=				num_audio_channels
        lowerCAmelCase__   :				Optional[int] 			=				hop_length
        lowerCAmelCase__   :				List[str] 			=				chunk_length
        lowerCAmelCase__   :				Optional[Any] 			=				sampling_rate






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        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			,					snake_case=False			,					snake_case=False   ):





        """simple docstring"""

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

        if equal_length:
         lowerCAmelCase__   :				Dict 			=				[floats_list((self.max_seq_length, self.feature_size)   ) for _ in range(self.batch_size   )]
        else:
         # make sure that inputs increase in size
         lowerCAmelCase__   :				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:
         lowerCAmelCase__   :				Dict 			=				[np.asarray(snake_case   ) for x in speech_inputs]
        return speech_inputs






@require_torch
@require_torchaudio
class 			__a					(						__magic_name__       ,	unittest.TestCase       ):





       """simple docstring"""


       __UpperCamelCase					:      Optional[Any]    =						TvltFeatureExtractor






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				Optional[Any] 			=				TvltFeatureExtractionTester(self   )






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				List[str] 			=				self.feature_extraction_class(**self.feat_extract_dict   )
        self.assertTrue(hasattr(snake_case			,					"spectrogram_length"   )   )
        self.assertTrue(hasattr(snake_case			,					"feature_size"   )   )
        self.assertTrue(hasattr(snake_case			,					"num_audio_channels"   )   )
        self.assertTrue(hasattr(snake_case			,					"hop_length"   )   )
        self.assertTrue(hasattr(snake_case			,					"chunk_length"   )   )
        self.assertTrue(hasattr(snake_case			,					"sampling_rate"   )   )






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				int 			=				self.feature_extraction_class(**self.feat_extract_dict   )

        with tempfile.TemporaryDirectory() as tmpdirname:
         lowerCAmelCase__   :				Union[str, Any] 			=				feat_extract_first.save_pretrained(snake_case   )[0]
         check_json_file_has_correct_format(snake_case   )
         lowerCAmelCase__   :				Tuple 			=				self.feature_extraction_class.from_pretrained(snake_case   )

        lowerCAmelCase__   :				Union[str, Any] 			=				feat_extract_first.to_dict()
        lowerCAmelCase__   :				int 			=				feat_extract_second.to_dict()
        lowerCAmelCase__   :				Union[str, Any] 			=				dict_first.pop("mel_filters"   )
        lowerCAmelCase__   :				List[Any] 			=				dict_second.pop("mel_filters"   )
        self.assertTrue(np.allclose(snake_case			,					snake_case   )   )
        self.assertEqual(snake_case			,					snake_case   )






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				Any 			=				self.feature_extraction_class(**self.feat_extract_dict   )

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

        lowerCAmelCase__   :				Optional[Any] 			=				feat_extract_first.to_dict()
        lowerCAmelCase__   :				Dict 			=				feat_extract_second.to_dict()
        lowerCAmelCase__   :				Tuple 			=				dict_first.pop("mel_filters"   )
        lowerCAmelCase__   :				Dict 			=				dict_second.pop("mel_filters"   )
        self.assertTrue(np.allclose(snake_case			,					snake_case   )   )
        self.assertEqual(snake_case			,					snake_case   )






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				Optional[int] 			=				self.feature_extraction_class(**self.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(800			,					1_400			,					200   )]
        lowerCAmelCase__   :				Optional[int] 			=				[np.asarray(snake_case   ) for speech_input in speech_inputs]

        # Test not batched input
        lowerCAmelCase__   :				Tuple 			=				feature_extractor(np_speech_inputs[0]			,					return_tensors="np"			,					sampling_rate=44_100   ).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
        lowerCAmelCase__   :				Optional[int] 			=				feature_extractor(snake_case			,					return_tensors="np"			,					sampling_rate=44_100   ).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
        lowerCAmelCase__   :				List[str] 			=				feature_extractor(
            snake_case			,					return_tensors="np"			,					sampling_rate=44_100			,					mask_audio=snake_case   ).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.
        lowerCAmelCase__   :				List[str] 			=				[floats_list((1, x)   )[0] for x in (800, 800, 800)]
        lowerCAmelCase__   :				Tuple 			=				np.asarray(snake_case   )
        lowerCAmelCase__   :				List[Any] 			=				feature_extractor(snake_case			,					return_tensors="np"			,					sampling_rate=44_100   ).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			,					snake_case   ):





        """simple docstring"""

        lowerCAmelCase__   :				str 			=				load_dataset("hf-internal-testing/librispeech_asr_dummy"			,					"clean"			,					split="validation"   )
        # automatic decoding with librispeech
        lowerCAmelCase__   :				int 			=				ds.sort("id"   ).select(range(snake_case   )   )[:num_samples]["audio"]

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






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				str 			=				self._load_datasamples(1   )
        lowerCAmelCase__   :				Optional[Any] 			=				TvltFeatureExtractor()
        lowerCAmelCase__   :				Dict 			=				feature_extractor(snake_case			,					return_tensors="pt"   ).audio_values

        self.assertEquals(audio_values.shape			,					(1, 1, 192, 128)   )

        lowerCAmelCase__   :				Dict 			=				torch.tensor([[-0.3_032, -0.2_708], [-0.4_434, -0.4_007]]   )
        self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]			,					snake_case			,					atol=1e-4   )   )






