
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 logging


__snake_case								=  logging.get_logger(__name__)
__snake_case								=  {"""vocab_file""": """spiece.model"""}

__snake_case								=  {
    """vocab_file""": {
        """albert-base-v1""": """https://huggingface.co/albert-base-v1/resolve/main/spiece.model""",
        """albert-large-v1""": """https://huggingface.co/albert-large-v1/resolve/main/spiece.model""",
        """albert-xlarge-v1""": """https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model""",
        """albert-xxlarge-v1""": """https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model""",
        """albert-base-v2""": """https://huggingface.co/albert-base-v2/resolve/main/spiece.model""",
        """albert-large-v2""": """https://huggingface.co/albert-large-v2/resolve/main/spiece.model""",
        """albert-xlarge-v2""": """https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model""",
        """albert-xxlarge-v2""": """https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model""",
    }
}

__snake_case								=  {
    """albert-base-v1""": 5_12,
    """albert-large-v1""": 5_12,
    """albert-xlarge-v1""": 5_12,
    """albert-xxlarge-v1""": 5_12,
    """albert-base-v2""": 5_12,
    """albert-large-v2""": 5_12,
    """albert-xlarge-v2""": 5_12,
    """albert-xxlarge-v2""": 5_12,
}

__snake_case								=  """▁"""




class 						lowercase__  (						snake_case__  ):
	A__    :   Tuple				       =VOCAB_FILES_NAMES
	A__    :   List[Any]				       =PRETRAINED_VOCAB_FILES_MAP
	A__    :   int				       =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	def __init__(			self							:   Dict       ,   UpperCAmelCase_							:   Optional[int]       ,   UpperCAmelCase_							:   Any=True       ,   UpperCAmelCase_							:   List[Any]=True       ,   UpperCAmelCase_							:   Optional[Any]=False       ,   UpperCAmelCase_							:   Any="[CLS]"       ,   UpperCAmelCase_							:   List[str]="[SEP]"       ,   UpperCAmelCase_							:   str="<unk>"       ,   UpperCAmelCase_							:   Any="[SEP]"       ,   UpperCAmelCase_							:   int="<pad>"       ,   UpperCAmelCase_							:   Union[str, Any]="[CLS]"       ,   UpperCAmelCase_							:   List[str]="[MASK]"       ,   UpperCAmelCase_							:   Optional[Dict[str, Any]] = None       ,   **UpperCAmelCase_							:   int       ,   ):
						# Mask token behave like a normal word, i.e. include the space before it and
						# is included in the raw text, there should be a match in a non-normalized sentence.
						SCREAMING_SNAKE_CASE__													= (
						    AddedToken(UpperCAmelCase_       ,   lstrip=UpperCAmelCase_       ,   rstrip=UpperCAmelCase_       ,   normalized=UpperCAmelCase_							)
						    if isinstance(UpperCAmelCase_       ,   UpperCAmelCase_							)
						    else mask_token
						)

						SCREAMING_SNAKE_CASE__													= {} 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_       ,   sp_model_kwargs=self.sp_model_kwargs       ,   **UpperCAmelCase_       ,   )

						SCREAMING_SNAKE_CASE__													= do_lower_case
						SCREAMING_SNAKE_CASE__													= remove_space
						SCREAMING_SNAKE_CASE__													= keep_accents
						SCREAMING_SNAKE_CASE__													= vocab_file

						SCREAMING_SNAKE_CASE__													= spm.SentencePieceProcessor(**self.sp_model_kwargs							)
						self.sp_model.Load(UpperCAmelCase_							)
	@property
	def      A_				(			self							:   str							):
						return len(self.sp_model							)
	def      A_				(			self							:   Any							):
						SCREAMING_SNAKE_CASE__													= {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							:   Optional[int]							):
						SCREAMING_SNAKE_CASE__													= self.__dict__.copy()
						SCREAMING_SNAKE_CASE__													= None
						return state
	def __setstate__(			self							:   List[str]       ,   UpperCAmelCase_							:   int							):
						SCREAMING_SNAKE_CASE__													= d

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

						SCREAMING_SNAKE_CASE__													= spm.SentencePieceProcessor(**self.sp_model_kwargs							)
						self.sp_model.Load(self.vocab_file							)
	def      A_				(			self							:   List[Any]       ,   UpperCAmelCase_							:   Optional[int]							):
						if self.remove_space:
											SCREAMING_SNAKE_CASE__													= " ".join(inputs.strip().split()							)
						else:
											SCREAMING_SNAKE_CASE__													= inputs
						SCREAMING_SNAKE_CASE__													= outputs.replace('``'       ,   '\"'							).replace('\'\''       ,   '\"'							)

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

						return outputs
	def      A_				(			self							:   str       ,   UpperCAmelCase_							:   str							):
						SCREAMING_SNAKE_CASE__													= self.preprocess_text(UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= self.sp_model.encode(UpperCAmelCase_       ,   out_type=UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= []
						for piece in pieces:
											if len(UpperCAmelCase_							) > 1 and piece[-1] == str(','							) and piece[-2].isdigit():
																SCREAMING_SNAKE_CASE__													= 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:
																										SCREAMING_SNAKE_CASE__													= cur_pieces[1:]
																					else:
																										SCREAMING_SNAKE_CASE__													= cur_pieces[0][1:]
																cur_pieces.append(piece[-1]							)
																new_pieces.extend(UpperCAmelCase_							)
											else:
																new_pieces.append(UpperCAmelCase_							)

						return new_pieces
	def      A_				(			self							:   Union[str, Any]       ,   UpperCAmelCase_							:   Optional[Any]							):
						return self.sp_model.PieceToId(UpperCAmelCase_							)
	def      A_				(			self							:   int       ,   UpperCAmelCase_							:   Optional[Any]							):
						return self.sp_model.IdToPiece(UpperCAmelCase_							)
	def      A_				(			self							:   int       ,   UpperCAmelCase_							:   Tuple							):
						SCREAMING_SNAKE_CASE__													= []
						SCREAMING_SNAKE_CASE__													= ""
						SCREAMING_SNAKE_CASE__													= 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(UpperCAmelCase_							) + token
																SCREAMING_SNAKE_CASE__													= True
																SCREAMING_SNAKE_CASE__													= []
											else:
																current_sub_tokens.append(UpperCAmelCase_							)
																SCREAMING_SNAKE_CASE__													= False
						out_string += self.sp_model.decode(UpperCAmelCase_							)
						return out_string.strip()
	def      A_				(			self							:   List[Any]       ,   UpperCAmelCase_							:   List[int]       ,   UpperCAmelCase_							:   Optional[List[int]] = None							):
						SCREAMING_SNAKE_CASE__													= [self.sep_token_id]
						SCREAMING_SNAKE_CASE__													= [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      A_				(			self							:   Tuple       ,   UpperCAmelCase_							:   List[int]       ,   UpperCAmelCase_							:   Optional[List[int]] = None       ,   UpperCAmelCase_							:   bool = 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 [1] + ([0] * len(UpperCAmelCase_							)) + [1] + ([0] * len(UpperCAmelCase_							)) + [1]
						return [1] + ([0] * len(UpperCAmelCase_							)) + [1]
	def      A_				(			self							:   int       ,   UpperCAmelCase_							:   List[int]       ,   UpperCAmelCase_							:   Optional[List[int]] = None							):
						SCREAMING_SNAKE_CASE__													= [self.sep_token_id]
						SCREAMING_SNAKE_CASE__													= [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      A_				(			self							:   Union[str, Any]       ,   UpperCAmelCase_							:   str       ,   UpperCAmelCase_							:   Optional[str] = None							):
						if not os.path.isdir(UpperCAmelCase_							):
											logger.error(F'Vocabulary path ({save_directory}) should be a directory'							)
											return
						SCREAMING_SNAKE_CASE__													= 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:
																SCREAMING_SNAKE_CASE__													= self.sp_model.serialized_model_proto()
																fi.write(UpperCAmelCase_							)

						return (out_vocab_file,)






'''simple docstring'''




import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_video_inputs


if is_torch_available():
  import torch

if is_vision_available():
  from PIL import Image

  from transformers import VivitImageProcessor

class 	lowerCamelCase_	(unittest.TestCase    ):
      '''simple docstring'''

      def __init__(       self						:				int						,					A						:				Dict						,					A						:				Optional[int]=7						,					A						:				Tuple=3						,					A						:				Optional[Any]=10						,					A						:				int=18						,					A						:				Dict=30						,					A						:				List[str]=400						,					A						:				int=True						,					A						:				Optional[Any]=None						,					A						:				Optional[Any]=True						,					A						:				List[Any]=[0.5, 0.5, 0.5]						,					A						:				List[str]=[0.5, 0.5, 0.5]						,					A						:				Optional[int]=None						,					):
             _UpperCAmelCase     :  Dict							      =    size if size is not None else {"shortest_edge": 18}
             _UpperCAmelCase     :  Optional[Any]							      =    crop_size if crop_size is not None else {"height": 18, "width": 18}

             _UpperCAmelCase     :  Tuple							      =    parent
             _UpperCAmelCase     :  Any							      =    batch_size
             _UpperCAmelCase     :  Optional[int]							      =    num_channels
             _UpperCAmelCase     :  Optional[Any]							      =    num_frames
             _UpperCAmelCase     :  Any							      =    image_size
             _UpperCAmelCase     :  Dict							      =    min_resolution
             _UpperCAmelCase     :  Any							      =    max_resolution
             _UpperCAmelCase     :  Optional[int]							      =    do_resize
             _UpperCAmelCase     :  str							      =    size
             _UpperCAmelCase     :  List[Any]							      =    do_normalize
             _UpperCAmelCase     :  Any							      =    image_mean
             _UpperCAmelCase     :  Tuple							      =    image_std
             _UpperCAmelCase     :  Any							      =    crop_size





      def      _A	(       self						:				List[Any]       ):
             return {
                 "image_mean": self.image_mean,
                 "image_std": self.image_std,
                 "do_normalize": self.do_normalize,
                 "do_resize": self.do_resize,
                 "size": self.size,
                 "crop_size": self.crop_size,
             }

@require_torch
@require_vision
class 	lowerCamelCase_	(snake_case__							, unittest.TestCase    ):
      '''simple docstring'''

      __UpperCamelCase:  Dict						=						VivitImageProcessor if is_vision_available() else None




      def      _A	(       self						:				int       ):
             _UpperCAmelCase     :  Tuple							      =    VivitImageProcessingTester(self       )




      @property
      def      _A	(       self						:				Optional[Any]       ):
             return self.image_processor_tester.prepare_image_processor_dict()




      def      _A	(       self						:				Union[str, Any]       ):
             _UpperCAmelCase     :  str							      =    self.image_processing_class(**self.image_processor_dict       )
             self.assertTrue(hasattr(A						,					"image_mean"       )       )
             self.assertTrue(hasattr(A						,					"image_std"       )       )
             self.assertTrue(hasattr(A						,					"do_normalize"       )       )
             self.assertTrue(hasattr(A						,					"do_resize"       )       )
             self.assertTrue(hasattr(A						,					"do_center_crop"       )       )
             self.assertTrue(hasattr(A						,					"size"       )       )




      def      _A	(       self						:				List[Any]       ):
             _UpperCAmelCase     :  Union[str, Any]							      =    self.image_processing_class.from_dict(self.image_processor_dict       )
             self.assertEqual(image_processor.size						,					{"shortest_edge": 18}       )
             self.assertEqual(image_processor.crop_size						,					{"height": 18, "width": 18}       )

             _UpperCAmelCase     :  Optional[int]							      =    self.image_processing_class.from_dict(self.image_processor_dict						,					size=42						,					crop_size=84       )
             self.assertEqual(image_processor.size						,					{"shortest_edge": 42}       )
             self.assertEqual(image_processor.crop_size						,					{"height": 84, "width": 84}       )




      def      _A	(       self						:				Tuple       ):
             # Initialize image_processing
             _UpperCAmelCase     :  Optional[Any]							      =    self.image_processing_class(**self.image_processor_dict       )
             # create random PIL videos
             _UpperCAmelCase     :  Any							      =    prepare_video_inputs(self.image_processor_tester						,					equal_resolution=A       )
             for video in video_inputs:
                    self.assertIsInstance(A						,					A       )
                    self.assertIsInstance(video[0]						,					Image.Image       )

             # Test not batched input
             _UpperCAmelCase     :  str							      =    image_processing(video_inputs[0]						,					return_tensors="pt"       ).pixel_values
             self.assertEqual(
                 encoded_videos.shape						,					(
                     1,
                     self.image_processor_tester.num_frames,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.crop_size["height"],
                     self.image_processor_tester.crop_size["width"],
                 )						,					)

             # Test batched
             _UpperCAmelCase     :  List[Any]							      =    image_processing(A						,					return_tensors="pt"       ).pixel_values
             self.assertEqual(
                 encoded_videos.shape						,					(
                     self.image_processor_tester.batch_size,
                     self.image_processor_tester.num_frames,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.crop_size["height"],
                     self.image_processor_tester.crop_size["width"],
                 )						,					)




      def      _A	(       self						:				List[Any]       ):
             # Initialize image_processing
             _UpperCAmelCase     :  List[str]							      =    self.image_processing_class(**self.image_processor_dict       )
             # create random numpy tensors
             _UpperCAmelCase     :  List[Any]							      =    prepare_video_inputs(self.image_processor_tester						,					equal_resolution=A						,					numpify=A       )
             for video in video_inputs:
                    self.assertIsInstance(A						,					A       )
                    self.assertIsInstance(video[0]						,					np.ndarray       )

             # Test not batched input
             _UpperCAmelCase     :  Tuple							      =    image_processing(video_inputs[0]						,					return_tensors="pt"       ).pixel_values
             self.assertEqual(
                 encoded_videos.shape						,					(
                     1,
                     self.image_processor_tester.num_frames,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.crop_size["height"],
                     self.image_processor_tester.crop_size["width"],
                 )						,					)

             # Test batched
             _UpperCAmelCase     :  Optional[int]							      =    image_processing(A						,					return_tensors="pt"       ).pixel_values
             self.assertEqual(
                 encoded_videos.shape						,					(
                     self.image_processor_tester.batch_size,
                     self.image_processor_tester.num_frames,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.crop_size["height"],
                     self.image_processor_tester.crop_size["width"],
                 )						,					)





      def      _A	(       self						:				List[Any]       ):
             # Initialize image_processing
             _UpperCAmelCase     :  Optional[Any]							      =    self.image_processing_class(**self.image_processor_dict       )
             # create random PyTorch tensors
             _UpperCAmelCase     :  Optional[int]							      =    prepare_video_inputs(self.image_processor_tester						,					equal_resolution=A						,					torchify=A       )
             for video in video_inputs:
                    self.assertIsInstance(A						,					A       )
                    self.assertIsInstance(video[0]						,					torch.Tensor       )

             # Test not batched input
             _UpperCAmelCase     :  Optional[Any]							      =    image_processing(video_inputs[0]						,					return_tensors="pt"       ).pixel_values
             self.assertEqual(
                 encoded_videos.shape						,					(
                     1,
                     self.image_processor_tester.num_frames,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.crop_size["height"],
                     self.image_processor_tester.crop_size["width"],
                 )						,					)

             # Test batched
             _UpperCAmelCase     :  List[Any]							      =    image_processing(A						,					return_tensors="pt"       ).pixel_values
             self.assertEqual(
                 encoded_videos.shape						,					(
                     self.image_processor_tester.batch_size,
                     self.image_processor_tester.num_frames,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.crop_size["height"],
                     self.image_processor_tester.crop_size["width"],
                 )						,					)

