



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


lowercase_								=     logging.get_logger(__name__)


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

# See all BART models at https://huggingface.co/models?filter=bart
lowercase_								=     {
    """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""",
    },
}

lowercase_								=     {
    """facebook/bart-base""": 1024,
    """facebook/bart-large""": 1024,
    """facebook/bart-large-mnli""": 1024,
    """facebook/bart-large-cnn""": 1024,
    """facebook/bart-large-xsum""": 1024,
    """yjernite/bart_eli5""": 1024,
}



@lru_cache()
def __lowerCAmelCase (     ):
			lowercase__					  =				(
			    list(range(ord("!"  )						,  ord("~"  ) + 1  )  ) + list(range(ord("¡"  )						,  ord("¬"  ) + 1  )  ) + list(range(ord("®"  )						,  ord("ÿ"  ) + 1  )  )
			)
			lowercase__					  =				bs[:]
			lowercase__					  =				0
			for b in range(2**8  ):
						if b not in bs:
									bs.append(SCREAMING_SNAKE_CASE_  )
									cs.append(2**8 + n  )
									n += 1
			lowercase__					  =				[chr(SCREAMING_SNAKE_CASE_  ) for n in cs]
			return dict(zip(SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_  )  )



def __lowerCAmelCase (     SCREAMING_SNAKE_CASE_  ):
			lowercase__					  =				set()
			lowercase__					  =				word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)  )
						lowercase__					  =				char
			return pairs




class 				_snake_case			(					lowercase__):
				UpperCamelCase__				:					Union[str, Any]     		=VOCAB_FILES_NAMES
				UpperCamelCase__				:					List[Any]     		=PRETRAINED_VOCAB_FILES_MAP
				UpperCamelCase__				:					str     		=PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				UpperCamelCase__				:					List[str]     		=["""input_ids""", """attention_mask"""]




				def __init__(						self					:						Optional[Any],					__lowercase					:						Dict,					__lowercase					:						Optional[int],					__lowercase					:						List[Any]="replace",					__lowercase					:						List[str]="<s>",					__lowercase					:						Optional[Any]="</s>",					__lowercase					:						int="</s>",					__lowercase					:						Optional[Any]="<s>",					__lowercase					:						str="<unk>",					__lowercase					:						Tuple="<pad>",					__lowercase					:						int="<mask>",					__lowercase					:						Dict=False,					**__lowercase					:						Optional[Any],					):
							lowercase__					  =				AddedToken(__lowercase,					lstrip=__lowercase,					rstrip=__lowercase						) if isinstance(__lowercase,					__lowercase						) else bos_token
							lowercase__					  =				AddedToken(__lowercase,					lstrip=__lowercase,					rstrip=__lowercase						) if isinstance(__lowercase,					__lowercase						) else eos_token
							lowercase__					  =				AddedToken(__lowercase,					lstrip=__lowercase,					rstrip=__lowercase						) if isinstance(__lowercase,					__lowercase						) else sep_token
							lowercase__					  =				AddedToken(__lowercase,					lstrip=__lowercase,					rstrip=__lowercase						) if isinstance(__lowercase,					__lowercase						) else cls_token
							lowercase__					  =				AddedToken(__lowercase,					lstrip=__lowercase,					rstrip=__lowercase						) if isinstance(__lowercase,					__lowercase						) else unk_token
							lowercase__					  =				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
							lowercase__					  =				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:
										lowercase__					  =				json.load(__lowercase						)
							lowercase__					  =				{v: k for k, v in self.encoder.items()}
							lowercase__					  =				errors  # how to handle errors in decoding
							lowercase__					  =				bytes_to_unicode()
							lowercase__					  =				{v: k for k, v in self.byte_encoder.items()}
							with open(__lowercase,					encoding="utf-8"						) as merges_handle:
										lowercase__					  =				merges_handle.read().split("\n"						)[1:-1]
							lowercase__					  =				[tuple(merge.split()						) for merge in bpe_merges]
							lowercase__					  =				dict(zip(__lowercase,					range(len(__lowercase						)						)						)						)
							lowercase__					  =				{}
							lowercase__					  =				add_prefix_space

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




				@property
				def 							A__      (						self					:						List[str]						):
							return len(self.encoder						)




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




				def 							A__      (						self					:						Optional[Any],					__lowercase					:						Dict						):
							if token in self.cache:
										return self.cache[token]
							lowercase__					  =				tuple(__lowercase						)
							lowercase__					  =				get_pairs(__lowercase						)

							if not pairs:
										return token

							while True:
										lowercase__					  =				min(__lowercase,					key=lambda __lowercase						: self.bpe_ranks.get(__lowercase,					float("inf"						)						)						)
										if bigram not in self.bpe_ranks:
													break
										lowercase__  ,       lowercase__					  =				bigram
										lowercase__					  =				[]
										lowercase__					  =				0
										while i < len(__lowercase						):
													try:
																lowercase__					  =				word.index(__lowercase,					__lowercase						)
													except ValueError:
																new_word.extend(word[i:]						)
																break
													else:
																new_word.extend(word[i:j]						)
																lowercase__					  =				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
										lowercase__					  =				tuple(__lowercase						)
										lowercase__					  =				new_word
										if len(__lowercase						) == 1:
													break
										else:
													lowercase__					  =				get_pairs(__lowercase						)
							lowercase__					  =				" ".join(__lowercase						)
							lowercase__					  =				word
							return word




				def 							A__      (						self					:						Tuple,					__lowercase					:						Dict						):
							lowercase__					  =				[]
							for token in re.findall(self.pat,					__lowercase						):
										lowercase__					  =				"".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 							A__      (						self					:						Union[str, Any],					__lowercase					:						Union[str, Any]						):
							return self.encoder.get(__lowercase,					self.encoder.get(self.unk_token						)						)




				def 							A__      (						self					:						Optional[Any],					__lowercase					:						int						):
							return self.decoder.get(__lowercase						)




				def 							A__      (						self					:						List[str],					__lowercase					:						Union[str, Any]						):
							lowercase__					  =				"".join(__lowercase						)
							lowercase__					  =				bytearray([self.byte_decoder[c] for c in text]						).decode("utf-8",					errors=self.errors						)
							return text




				def 							A__      (						self					:						Tuple,					__lowercase					:						str,					__lowercase					:						Optional[str] = None						):
							if not os.path.isdir(__lowercase						):
										logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''						)
										return
							lowercase__					  =				os.path.join(
							    __lowercase,					(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]						)
							lowercase__					  =				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"						)

							lowercase__					  =				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 __lowercase						: 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!"						)
																lowercase__					  =				token_index
													writer.write(" ".join(__lowercase						) + "\n"						)
													index += 1

							return vocab_file, merge_file




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




				def 							A__      (						self					:						Any,					__lowercase					:						List[int],					__lowercase					:						Optional[List[int]] = None,					__lowercase					:						bool = 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 							A__      (						self					:						str,					__lowercase					:						List[int],					__lowercase					:						Optional[List[int]] = None						):
							lowercase__					  =				[self.sep_token_id]
							lowercase__					  =				[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 							A__      (						self					:						Optional[Any],					__lowercase					:						Any,					__lowercase					:						Dict=False,					**__lowercase					:						Union[str, Any]						):
							lowercase__					  =				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()):
										lowercase__					  =				" " + text
							return (text, kwargs)









import unittest

import numpy as np
import requests

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_image_inputs


if is_torch_available():
							import torch

							from transformers.pytorch_utils import is_torch_greater_or_equal_than_1_11
else:
							lowercase_								=     False

if is_vision_available():
							from PIL import Image

							from transformers import PixaStructImageProcessor
class 				_snake_case			(					unittest.TestCase):




				def __init__(						self					:						List[Any],					__lowercase					:						int,					__lowercase					:						Optional[int]=7,					__lowercase					:						List[str]=3,					__lowercase					:						Tuple=18,					__lowercase					:						List[Any]=30,					__lowercase					:						Tuple=400,					__lowercase					:						Any=None,					__lowercase					:						Optional[int]=True,					__lowercase					:						List[str]=True,					__lowercase					:						Union[str, Any]=None,					):
							lowercase__					  =				size if size is not None else {"height": 20, "width": 20}
							lowercase__					  =				parent
							lowercase__					  =				batch_size
							lowercase__					  =				num_channels
							lowercase__					  =				image_size
							lowercase__					  =				min_resolution
							lowercase__					  =				max_resolution
							lowercase__					  =				size
							lowercase__					  =				do_normalize
							lowercase__					  =				do_convert_rgb
							lowercase__					  =				[512, 1024, 2048, 4096]
							lowercase__					  =				patch_size if patch_size is not None else {"height": 16, "width": 16}




				def 							A__      (						self					:						List[str]						):
							return {"do_normalize": self.do_normalize, "do_convert_rgb": self.do_convert_rgb}




				def 							A__      (						self					:						Any						):
							lowercase__					  =				"https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/australia.jpg"
							lowercase__					  =				Image.open(requests.get(__lowercase,					stream=__lowercase						).raw						).convert("RGB"						)
							return raw_image
@unittest.skipIf(
    not is_torch_greater_or_equal_than_1_11    ,	reason="""`Pix2StructImageProcessor` requires `torch>=1.11.0`."""    ,	)
@require_torch
@require_vision
class 				_snake_case			(					lowercase__    ,	unittest.TestCase):
				UpperCamelCase__				:					Any     		=PixaStructImageProcessor if is_vision_available() else None




				def 							A__      (						self					:						Any						):
							lowercase__					  =				PixaStructImageProcessingTester(self						)




				@property
				def 							A__      (						self					:						Union[str, Any]						):
							return self.image_processor_tester.prepare_image_processor_dict()




				def 							A__      (						self					:						Optional[Any]						):
							lowercase__					  =				self.image_processing_class(**self.image_processor_dict						)
							self.assertTrue(hasattr(__lowercase,					"do_normalize"						)						)
							self.assertTrue(hasattr(__lowercase,					"do_convert_rgb"						)						)




				def 							A__      (						self					:						Optional[int]						):
							lowercase__					  =				self.image_processor_tester.prepare_dummy_image()

							lowercase__					  =				self.image_processing_class(**self.image_processor_dict						)
							lowercase__					  =				2048

							lowercase__					  =				image_processor(__lowercase,					return_tensors="pt",					max_patches=__lowercase						)
							self.assertTrue(torch.allclose(inputs.flattened_patches.mean(),					torch.tensor(0.0606						),					atol=1e-3,					rtol=1e-3						)						)




				def 							A__      (						self					:						Union[str, Any]						):
							# Initialize image_processor
							lowercase__					  =				self.image_processing_class(**self.image_processor_dict						)
							# create random PIL images
							lowercase__					  =				prepare_image_inputs(self.image_processor_tester,					equal_resolution=__lowercase						)
							for image in image_inputs:
										self.assertIsInstance(__lowercase,					Image.Image						)

							# Test not batched input
							lowercase__					  =				(
							    (self.image_processor_tester.patch_size["height"] * self.image_processor_tester.patch_size["width"])
							    * self.image_processor_tester.num_channels
							) + 2

							for max_patch in self.image_processor_tester.max_patches:
										# Test not batched input
										lowercase__					  =				image_processor(
										    image_inputs[0],					return_tensors="pt",					max_patches=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(1, max_patch, expected_hidden_dim),					)

										# Test batched
										lowercase__					  =				image_processor(
										    __lowercase,					return_tensors="pt",					max_patches=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(self.image_processor_tester.batch_size, max_patch, expected_hidden_dim),					)




				def 							A__      (						self					:						int						):
							# Initialize image_processor
							lowercase__					  =				self.image_processing_class(**self.image_processor_dict						)
							# create random PIL images
							lowercase__					  =				prepare_image_inputs(self.image_processor_tester,					equal_resolution=__lowercase						)
							for image in image_inputs:
										self.assertIsInstance(__lowercase,					Image.Image						)

							# Test not batched input
							lowercase__					  =				(
							    (self.image_processor_tester.patch_size["height"] * self.image_processor_tester.patch_size["width"])
							    * self.image_processor_tester.num_channels
							) + 2

							lowercase__					  =				True

							for max_patch in self.image_processor_tester.max_patches:
										# Test not batched input
										with self.assertRaises(__lowercase						):
													lowercase__					  =				image_processor(
													    image_inputs[0],					return_tensors="pt",					max_patches=__lowercase						).flattened_patches

										lowercase__					  =				"Hello"

										lowercase__					  =				image_processor(
										    image_inputs[0],					return_tensors="pt",					max_patches=__lowercase,					header_text=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(1, max_patch, expected_hidden_dim),					)

										# Test batched
										lowercase__					  =				image_processor(
										    __lowercase,					return_tensors="pt",					max_patches=__lowercase,					header_text=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(self.image_processor_tester.batch_size, max_patch, expected_hidden_dim),					)




				def 							A__      (						self					:						Tuple						):
							# Initialize image_processor
							lowercase__					  =				self.image_processing_class(**self.image_processor_dict						)
							# create random numpy tensors
							lowercase__					  =				prepare_image_inputs(self.image_processor_tester,					equal_resolution=__lowercase,					numpify=__lowercase						)
							for image in image_inputs:
										self.assertIsInstance(__lowercase,					np.ndarray						)

							lowercase__					  =				(
							    (self.image_processor_tester.patch_size["height"] * self.image_processor_tester.patch_size["width"])
							    * self.image_processor_tester.num_channels
							) + 2

							for max_patch in self.image_processor_tester.max_patches:
										# Test not batched input
										lowercase__					  =				image_processor(
										    image_inputs[0],					return_tensors="pt",					max_patches=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(1, max_patch, expected_hidden_dim),					)

										# Test batched
										lowercase__					  =				image_processor(
										    __lowercase,					return_tensors="pt",					max_patches=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(self.image_processor_tester.batch_size, max_patch, expected_hidden_dim),					)




				def 							A__      (						self					:						Any						):
							# Initialize image_processor
							lowercase__					  =				self.image_processing_class(**self.image_processor_dict						)
							# create random PyTorch tensors
							lowercase__					  =				prepare_image_inputs(self.image_processor_tester,					equal_resolution=__lowercase,					torchify=__lowercase						)
							for image in image_inputs:
										self.assertIsInstance(__lowercase,					torch.Tensor						)

							# Test not batched input
							lowercase__					  =				(
							    (self.image_processor_tester.patch_size["height"] * self.image_processor_tester.patch_size["width"])
							    * self.image_processor_tester.num_channels
							) + 2

							for max_patch in self.image_processor_tester.max_patches:
										# Test not batched input
										lowercase__					  =				image_processor(
										    image_inputs[0],					return_tensors="pt",					max_patches=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(1, max_patch, expected_hidden_dim),					)

										# Test batched
										lowercase__					  =				image_processor(
										    __lowercase,					return_tensors="pt",					max_patches=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(self.image_processor_tester.batch_size, max_patch, expected_hidden_dim),					)
@unittest.skipIf(
    not is_torch_greater_or_equal_than_1_11    ,	reason="""`Pix2StructImageProcessor` requires `torch>=1.11.0`."""    ,	)
@require_torch
@require_vision
class 				_snake_case			(					lowercase__    ,	unittest.TestCase):
				UpperCamelCase__				:					Optional[int]     		=PixaStructImageProcessor if is_vision_available() else None




				def 							A__      (						self					:						Optional[int]						):
							lowercase__					  =				PixaStructImageProcessingTester(self,					num_channels=4						)
							lowercase__					  =				3




				@property
				def 							A__      (						self					:						Union[str, Any]						):
							return self.image_processor_tester.prepare_image_processor_dict()




				def 							A__      (						self					:						Dict						):
							lowercase__					  =				self.image_processing_class(**self.image_processor_dict						)
							self.assertTrue(hasattr(__lowercase,					"do_normalize"						)						)
							self.assertTrue(hasattr(__lowercase,					"do_convert_rgb"						)						)




				def 							A__      (						self					:						Union[str, Any]						):
							# Initialize image_processor
							lowercase__					  =				self.image_processing_class(**self.image_processor_dict						)
							# create random PIL images
							lowercase__					  =				prepare_image_inputs(self.image_processor_tester,					equal_resolution=__lowercase						)
							for image in image_inputs:
										self.assertIsInstance(__lowercase,					Image.Image						)

							# Test not batched input
							lowercase__					  =				(
							    (self.image_processor_tester.patch_size["height"] * self.image_processor_tester.patch_size["width"])
							    * (self.image_processor_tester.num_channels - 1)
							) + 2

							for max_patch in self.image_processor_tester.max_patches:
										# Test not batched input
										lowercase__					  =				image_processor(
										    image_inputs[0],					return_tensors="pt",					max_patches=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(1, max_patch, expected_hidden_dim),					)

										# Test batched
										lowercase__					  =				image_processor(
										    __lowercase,					return_tensors="pt",					max_patches=__lowercase						).flattened_patches
										self.assertEqual(
										    encoded_images.shape,					(self.image_processor_tester.batch_size, max_patch, expected_hidden_dim),					)




