



import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

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


UpperCAmelCase_									=     logging.get_logger(__name__)


UpperCAmelCase_									=     {"vocab_file": "vocab.json", "merges_file": "merges.txt"}

# See all BART models at https://huggingface.co/models?filter=bart
UpperCAmelCase_									=     {
    "vocab_file": {
        "facebook/bart-base": "https://huggingface.co/facebook/bart-base/resolve/main/vocab.json",
        "facebook/bart-large": "https://huggingface.co/facebook/bart-large/resolve/main/vocab.json",
        "facebook/bart-large-mnli": "https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json",
        "facebook/bart-large-cnn": "https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json",
        "facebook/bart-large-xsum": "https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json",
        "yjernite/bart_eli5": "https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json",
    },
    "merges_file": {
        "facebook/bart-base": "https://huggingface.co/facebook/bart-base/resolve/main/merges.txt",
        "facebook/bart-large": "https://huggingface.co/facebook/bart-large/resolve/main/merges.txt",
        "facebook/bart-large-mnli": "https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt",
        "facebook/bart-large-cnn": "https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt",
        "facebook/bart-large-xsum": "https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt",
        "yjernite/bart_eli5": "https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt",
    },
}

UpperCAmelCase_									=     {
    "facebook/bart-base": 1_0_2_4,
    "facebook/bart-large": 1_0_2_4,
    "facebook/bart-large-mnli": 1_0_2_4,
    "facebook/bart-large-cnn": 1_0_2_4,
    "facebook/bart-large-xsum": 1_0_2_4,
    "yjernite/bart_eli5": 1_0_2_4,
}
@lru_cache()
def 							UpperCAmelCase__		(				)->Union[str, Any]:
      _lowerCAmelCase 			=  (
          list(range(ord('''!'''				)     ,				ord('''~'''				) + 1				)				) + list(range(ord('''¡'''				)     ,				ord('''¬'''				) + 1				)				) + list(range(ord('''®'''				)     ,				ord('''ÿ'''				) + 1				)				)
      )
      _lowerCAmelCase 			=  bs[:]
      _lowerCAmelCase 			=  0
      for b in range(2**8				):
            if b not in bs:
                  bs.append(SCREAMING_SNAKE_CASE_				)
                  cs.append(2**8 + n				)
                  n += 1
      _lowerCAmelCase 			=  [chr(SCREAMING_SNAKE_CASE_				) for n in cs]
      return dict(zip(SCREAMING_SNAKE_CASE_     ,				SCREAMING_SNAKE_CASE_				)				)
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Tuple				)->Union[str, Any]:
      _lowerCAmelCase 			=  set()
      _lowerCAmelCase 			=  word[0]
      for char in word[1:]:
            pairs.add((prev_char, char)				)
            _lowerCAmelCase 			=  char
      return pairs





class 		UpperCAmelCase						(						lowercase__				):
   SCREAMING_SNAKE_CASE__							=     VOCAB_FILES_NAMES
   SCREAMING_SNAKE_CASE__							=     PRETRAINED_VOCAB_FILES_MAP
   SCREAMING_SNAKE_CASE__							=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   SCREAMING_SNAKE_CASE__							=     ["""input_ids""", """attention_mask"""]





   def __init__(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase       ,     _lowerCAmelCase="replace"       ,     _lowerCAmelCase="<s>"       ,     _lowerCAmelCase="</s>"       ,     _lowerCAmelCase="</s>"       ,     _lowerCAmelCase="<s>"       ,     _lowerCAmelCase="<unk>"       ,     _lowerCAmelCase="<pad>"       ,     _lowerCAmelCase="<mask>"       ,     _lowerCAmelCase=False       ,     **_lowerCAmelCase       ,     ):
         _lowerCAmelCase 			=  AddedToken(__lowercase       ,     lstrip=__lowercase       ,     rstrip=__lowercase			) if isinstance(__lowercase       ,     __lowercase			) else bos_token
         _lowerCAmelCase 			=  AddedToken(__lowercase       ,     lstrip=__lowercase       ,     rstrip=__lowercase			) if isinstance(__lowercase       ,     __lowercase			) else eos_token
         _lowerCAmelCase 			=  AddedToken(__lowercase       ,     lstrip=__lowercase       ,     rstrip=__lowercase			) if isinstance(__lowercase       ,     __lowercase			) else sep_token
         _lowerCAmelCase 			=  AddedToken(__lowercase       ,     lstrip=__lowercase       ,     rstrip=__lowercase			) if isinstance(__lowercase       ,     __lowercase			) else cls_token
         _lowerCAmelCase 			=  AddedToken(__lowercase       ,     lstrip=__lowercase       ,     rstrip=__lowercase			) if isinstance(__lowercase       ,     __lowercase			) else unk_token
         _lowerCAmelCase 			=  AddedToken(__lowercase       ,     lstrip=__lowercase       ,     rstrip=__lowercase			) if isinstance(__lowercase       ,     __lowercase			) else pad_token

         # Mask token behave like a normal word, i.e. include the space before it
         _lowerCAmelCase 			=  AddedToken(__lowercase       ,     lstrip=__lowercase       ,     rstrip=__lowercase			) if isinstance(__lowercase       ,     __lowercase			) else mask_token

         super().__init__(
             errors=__lowercase       ,     bos_token=__lowercase       ,     eos_token=__lowercase       ,     unk_token=__lowercase       ,     sep_token=__lowercase       ,     cls_token=__lowercase       ,     pad_token=__lowercase       ,     mask_token=__lowercase       ,     add_prefix_space=__lowercase       ,     **__lowercase       ,     )

         with open(__lowercase       ,     encoding='''utf-8'''			) as vocab_handle:
               _lowerCAmelCase 			=  json.load(__lowercase			)
         _lowerCAmelCase 			=  {v: k for k, v in self.encoder.items()}
         _lowerCAmelCase 			=  errors  # how to handle errors in decoding
         _lowerCAmelCase 			=  bytes_to_unicode()
         _lowerCAmelCase 			=  {v: k for k, v in self.byte_encoder.items()}
         with open(__lowercase       ,     encoding='''utf-8'''			) as merges_handle:
               _lowerCAmelCase 			=  merges_handle.read().split('''\n'''			)[1:-1]
         _lowerCAmelCase 			=  [tuple(merge.split()			) for merge in bpe_merges]
         _lowerCAmelCase 			=  dict(zip(__lowercase       ,     range(len(__lowercase			)			)			)			)
         _lowerCAmelCase 			=  {}
         _lowerCAmelCase 			=  add_prefix_space

         # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
         _lowerCAmelCase 			=  re.compile(R'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''			)





   @property
   def 	__lowerCAmelCase							(	self			):
         return len(self.encoder			)





   def 	__lowerCAmelCase							(	self			):
         return dict(self.encoder       ,     **self.added_tokens_encoder			)





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase			):
         if token in self.cache:
               return self.cache[token]
         _lowerCAmelCase 			=  tuple(__lowercase			)
         _lowerCAmelCase 			=  get_pairs(__lowercase			)

         if not pairs:
               return token

         while True:
               _lowerCAmelCase 			=  min(__lowercase       ,     key=lambda _lowerCAmelCase			: self.bpe_ranks.get(__lowercase       ,     float('''inf'''			)			)			)
               if bigram not in self.bpe_ranks:
                     break
               _lowerCAmelCase			,   _lowerCAmelCase 			=  bigram
               _lowerCAmelCase 			=  []
               _lowerCAmelCase 			=  0
               while i < len(__lowercase			):
                     try:
                           _lowerCAmelCase 			=  word.index(__lowercase       ,     __lowercase			)
                     except ValueError:
                           new_word.extend(word[i:]			)
                           break
                     else:
                           new_word.extend(word[i:j]			)
                           _lowerCAmelCase 			=  j

                     if word[i] == first and i < len(__lowercase			) - 1 and word[i + 1] == second:
                           new_word.append(first + second			)
                           i += 2
                     else:
                           new_word.append(word[i]			)
                           i += 1
               _lowerCAmelCase 			=  tuple(__lowercase			)
               _lowerCAmelCase 			=  new_word
               if len(__lowercase			) == 1:
                     break
               else:
                     _lowerCAmelCase 			=  get_pairs(__lowercase			)
         _lowerCAmelCase 			=  ''' '''.join(__lowercase			)
         _lowerCAmelCase 			=  word
         return word





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase			):
         _lowerCAmelCase 			=  []
         for token in re.findall(self.pat       ,     __lowercase			):
               _lowerCAmelCase 			=  ''''''.join(
                   self.byte_encoder[b] for b in token.encode('''utf-8'''			)			)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
               bpe_tokens.extend(bpe_token for bpe_token in self.bpe(__lowercase			).split(''' '''			)			)
         return bpe_tokens





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase			):
         return self.encoder.get(__lowercase       ,     self.encoder.get(self.unk_token			)			)





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase			):
         return self.decoder.get(__lowercase			)





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase			):
         _lowerCAmelCase 			=  ''''''.join(__lowercase			)
         _lowerCAmelCase 			=  bytearray([self.byte_decoder[c] for c in text]			).decode('''utf-8'''       ,     errors=self.errors			)
         return text





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase = None			):
         if not os.path.isdir(__lowercase			):
               logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''			)
               return
         _lowerCAmelCase 			=  os.path.join(
             __lowercase       ,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']			)
         _lowerCAmelCase 			=  os.path.join(
             __lowercase       ,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']			)

         with open(__lowercase       ,     '''w'''       ,     encoding='''utf-8'''			) as f:
               f.write(json.dumps(self.encoder       ,     indent=2       ,     sort_keys=__lowercase       ,     ensure_ascii=__lowercase			) + '''\n'''			)

         _lowerCAmelCase 			=  0
         with open(__lowercase       ,     '''w'''       ,     encoding='''utf-8'''			) as writer:
               writer.write('''#version: 0.2\n'''			)
               for bpe_tokens, token_index in sorted(self.bpe_ranks.items()       ,     key=lambda _lowerCAmelCase			: kv[1]			):
                     if index != token_index:
                           logger.warning(
                               F'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
                               ''' Please check that the tokenizer is not corrupted!'''			)
                           _lowerCAmelCase 			=  token_index
                     writer.write(''' '''.join(__lowercase			) + '''\n'''			)
                     index += 1

         return vocab_file, merge_file





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase = None			):
         if token_ids_a is None:
               return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
         _lowerCAmelCase 			=  [self.cls_token_id]
         _lowerCAmelCase 			=  [self.sep_token_id]
         return cls + token_ids_a + sep + sep + token_ids_a + sep





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase = None       ,     _lowerCAmelCase = False			):
         if already_has_special_tokens:
               return super().get_special_tokens_mask(
                   token_ids_a=__lowercase       ,     token_ids_a=__lowercase       ,     already_has_special_tokens=__lowercase			)

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





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase = None			):
         _lowerCAmelCase 			=  [self.sep_token_id]
         _lowerCAmelCase 			=  [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 + sep + token_ids_a + sep			) * [0]





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase=False       ,     **_lowerCAmelCase			):
         _lowerCAmelCase 			=  kwargs.pop('''add_prefix_space'''       ,     self.add_prefix_space			)
         if (is_split_into_words or add_prefix_space) and (len(__lowercase			) > 0 and not text[0].isspace()):
               _lowerCAmelCase 			=  ''' ''' + text
         return (text, kwargs)




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 		UpperCAmelCase						(						unittest.TestCase				):





   def __init__(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase=7       ,     _lowerCAmelCase=3       ,     _lowerCAmelCase=10       ,     _lowerCAmelCase=18       ,     _lowerCAmelCase=30       ,     _lowerCAmelCase=400       ,     _lowerCAmelCase=True       ,     _lowerCAmelCase=None       ,     _lowerCAmelCase=True       ,     _lowerCAmelCase=[0.5, 0.5, 0.5]       ,     _lowerCAmelCase=[0.5, 0.5, 0.5]       ,     _lowerCAmelCase=None       ,     ):
         _lowerCAmelCase 			=  size if size is not None else {'''shortest_edge''': 18}
         _lowerCAmelCase 			=  crop_size if crop_size is not None else {'''height''': 18, '''width''': 18}

         _lowerCAmelCase 			=  parent
         _lowerCAmelCase 			=  batch_size
         _lowerCAmelCase 			=  num_channels
         _lowerCAmelCase 			=  num_frames
         _lowerCAmelCase 			=  image_size
         _lowerCAmelCase 			=  min_resolution
         _lowerCAmelCase 			=  max_resolution
         _lowerCAmelCase 			=  do_resize
         _lowerCAmelCase 			=  size
         _lowerCAmelCase 			=  do_normalize
         _lowerCAmelCase 			=  image_mean
         _lowerCAmelCase 			=  image_std
         _lowerCAmelCase 			=  crop_size





   def 	__lowerCAmelCase							(	self			):
         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 		UpperCAmelCase						(						snake_case_       ,unittest.TestCase				):
   SCREAMING_SNAKE_CASE__							=     VivitImageProcessor if is_vision_available() else None





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  VivitImageProcessingTester(self			)





   @property
   def 	__lowerCAmelCase							(	self			):
         return self.image_processor_tester.prepare_image_processor_dict()





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''image_mean'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''image_std'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''do_normalize'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''do_resize'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''do_center_crop'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''size'''			)			)





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  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}			)

         _lowerCAmelCase 			=  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 	__lowerCAmelCase							(	self			):
         # Initialize image_processing
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         # create random PIL videos
         _lowerCAmelCase 			=  prepare_video_inputs(self.image_processor_tester       ,     equal_resolution=_lowerCAmelCase			)
         for video in video_inputs:
               self.assertIsInstance(_lowerCAmelCase       ,     _lowerCAmelCase			)
               self.assertIsInstance(video[0]       ,     Image.Image			)

         # Test not batched input
         _lowerCAmelCase 			=  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
         _lowerCAmelCase 			=  image_processing(_lowerCAmelCase       ,     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 	__lowerCAmelCase							(	self			):
         # Initialize image_processing
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         # create random numpy tensors
         _lowerCAmelCase 			=  prepare_video_inputs(self.image_processor_tester       ,     equal_resolution=_lowerCAmelCase       ,     numpify=_lowerCAmelCase			)
         for video in video_inputs:
               self.assertIsInstance(_lowerCAmelCase       ,     _lowerCAmelCase			)
               self.assertIsInstance(video[0]       ,     np.ndarray			)

         # Test not batched input
         _lowerCAmelCase 			=  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
         _lowerCAmelCase 			=  image_processing(_lowerCAmelCase       ,     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 	__lowerCAmelCase							(	self			):
         # Initialize image_processing
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         # create random PyTorch tensors
         _lowerCAmelCase 			=  prepare_video_inputs(self.image_processor_tester       ,     equal_resolution=_lowerCAmelCase       ,     torchify=_lowerCAmelCase			)
         for video in video_inputs:
               self.assertIsInstance(_lowerCAmelCase       ,     _lowerCAmelCase			)
               self.assertIsInstance(video[0]       ,     torch.Tensor			)

         # Test not batched input
         _lowerCAmelCase 			=  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
         _lowerCAmelCase 			=  image_processing(_lowerCAmelCase       ,     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'''],
             )       ,     )