



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

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


UpperCAmelCase__     = logging.get_logger(__name__)


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

# See all LED models at https://huggingface.co/models?filter=LED
UpperCAmelCase__     = {
    "vocab_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json",
    },
    "merges_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt",
    },
    "tokenizer_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json",
    },
}

UpperCAmelCase__     = {
    "allenai/led-base-16384": 1_6384,
}





@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def 							A			(  )       ->				Dict:







			'''simple docstring'''
			_UpperCAmelCase	 =     (
			    list(range(ord('!'					)							,     ord('~'					) + 1					)					) + list(range(ord('¡'					)							,     ord('¬'					) + 1					)					) + list(range(ord('®'					)							,     ord('ÿ'					) + 1					)					)
			)
			_UpperCAmelCase	 =     bs[:]
			_UpperCAmelCase	 =     0
			for b in range(2**8					):
						if b not in bs:
									bs.append(_UpperCAmelCase					)
									cs.append(2**8 + n					)
									n += 1
			_UpperCAmelCase	 =     [chr(_UpperCAmelCase					) for n in cs]
			return dict(zip(_UpperCAmelCase							,     _UpperCAmelCase					)					)





def 							A			(  _UpperCAmelCase       :     Union[str, Any]					)       ->				Dict:







			'''simple docstring'''
			_UpperCAmelCase	 =     set()
			_UpperCAmelCase	 =     word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)					)
						_UpperCAmelCase	 =     char
			return pairs







class     __lowerCAmelCase					(    A		):
			UpperCamelCase           =  VOCAB_FILES_NAMES
			UpperCamelCase           =  PRETRAINED_VOCAB_FILES_MAP
			UpperCamelCase           =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			UpperCamelCase           =  ['''input_ids''', '''attention_mask''']



			def __init__(		self     :						Union[str, Any]				,	A     :						int				,	A     :						Any				,	A     :						Tuple="replace"				,	A     :						Any="<s>"				,	A     :						List[Any]="</s>"				,	A     :						int="</s>"				,	A     :						List[Any]="<s>"				,	A     :						Any="<unk>"				,	A     :						Tuple="<pad>"				,	A     :						Optional[Any]="<mask>"				,	A     :						str=False				,	**A     :						Union[str, Any]				,	)     ->					Optional[Any]:



						"""simple docstring"""






						_UpperCAmelCase	 =     AddedToken(A				,	lstrip=A				,	rstrip=A) if isinstance(A				,	A) else bos_token
						_UpperCAmelCase	 =     AddedToken(A				,	lstrip=A				,	rstrip=A) if isinstance(A				,	A) else eos_token
						_UpperCAmelCase	 =     AddedToken(A				,	lstrip=A				,	rstrip=A) if isinstance(A				,	A) else sep_token
						_UpperCAmelCase	 =     AddedToken(A				,	lstrip=A				,	rstrip=A) if isinstance(A				,	A) else cls_token
						_UpperCAmelCase	 =     AddedToken(A				,	lstrip=A				,	rstrip=A) if isinstance(A				,	A) else unk_token
						_UpperCAmelCase	 =     AddedToken(A				,	lstrip=A				,	rstrip=A) if isinstance(A				,	A) else pad_token

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

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

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

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



			@property
			# Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
			def 					_lowerCamelCase (		self     :						str)     ->					Optional[Any]:



						"""simple docstring"""






						return len(self.encoder)



			def 					_lowerCamelCase (		self     :						Tuple)     ->					Optional[Any]:



						"""simple docstring"""






						return dict(self.encoder				,	**self.added_tokens_encoder)



			def 					_lowerCamelCase (		self     :						Optional[Any]				,	A     :						Dict)     ->					List[Any]:



						"""simple docstring"""






						if token in self.cache:
									return self.cache[token]
						_UpperCAmelCase	 =     tuple(A)
						_UpperCAmelCase	 =     get_pairs(A)

						if not pairs:
									return token

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

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



			def 					_lowerCamelCase (		self     :						Optional[int]				,	A     :						Optional[int])     ->					int:



						"""simple docstring"""






						_UpperCAmelCase	 =     []
						for token in re.findall(self.pat				,	A):
									_UpperCAmelCase	 =     ''.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(A).split(' '))
						return bpe_tokens



			def 					_lowerCamelCase (		self     :						Any				,	A     :						Optional[Any])     ->					Union[str, Any]:



						"""simple docstring"""






						return self.encoder.get(A				,	self.encoder.get(self.unk_token))



			def 					_lowerCamelCase (		self     :						str				,	A     :						int)     ->					Union[str, Any]:



						"""simple docstring"""






						return self.decoder.get(A)



			def 					_lowerCamelCase (		self     :						List[Any]				,	A     :						int)     ->					Optional[Any]:



						"""simple docstring"""






						_UpperCAmelCase	 =     ''.join(A)
						_UpperCAmelCase	 =     bytearray([self.byte_decoder[c] for c in text]).decode('utf-8'				,	errors=self.errors)
						return text



			def 					_lowerCamelCase (		self     :						Any				,	A     :						str				,	A     :						Optional[str] = None)     ->					Tuple[str]:



						"""simple docstring"""






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

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

						_UpperCAmelCase	 =     0
						with open(A				,	'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 A: 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!')
															_UpperCAmelCase	 =     token_index
												writer.write(' '.join(A) + '\n')
												index += 1

						return vocab_file, merge_file



			def 					_lowerCamelCase (		self     :						Dict				,	A     :						List[int]				,	A     :						Optional[List[int]] = None)     ->					List[int]:



						"""simple docstring"""






						if token_ids_a is None:
									return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
						_UpperCAmelCase	 =     [self.cls_token_id]
						_UpperCAmelCase	 =     [self.sep_token_id]
						return cls + token_ids_a + sep + sep + token_ids_a + sep



			def 					_lowerCamelCase (		self     :						str				,	A     :						List[int]				,	A     :						Optional[List[int]] = None				,	A     :						bool = False)     ->					List[int]:



						"""simple docstring"""






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

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



			def 					_lowerCamelCase (		self     :						List[str]				,	A     :						List[int]				,	A     :						Optional[List[int]] = None)     ->					List[int]:



						"""simple docstring"""






						_UpperCAmelCase	 =     [self.sep_token_id]
						_UpperCAmelCase	 =     [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     :						Any				,	A     :						Optional[int]				,	A     :						str=False				,	**A     :						List[Any])     ->					Dict:



						"""simple docstring"""






						_UpperCAmelCase	 =     kwargs.pop('add_prefix_space'				,	self.add_prefix_space)
						if (is_split_into_words or add_prefix_space) and (len(A) > 0 and not text[0].isspace()):
									_UpperCAmelCase	 =     ' ' + text
						return (text, kwargs)



			def 					_lowerCamelCase (		self     :						List[Any]				,	A     :						Union[Dict[str, EncodedInput], BatchEncoding]				,	A     :						Optional[int] = None				,	A     :						PaddingStrategy = PaddingStrategy.DO_NOT_PAD				,	A     :						Optional[int] = None				,	A     :						Optional[bool] = None				,	)     ->					dict:



						"""simple docstring"""






						_UpperCAmelCase	 =     super()._pad(
						    encoded_inputs=A				,	max_length=A				,	padding_strategy=A				,	pad_to_multiple_of=A				,	return_attention_mask=A				,	)

						# Load from model defaults
						if return_attention_mask is None:
									_UpperCAmelCase	 =     'attention_mask' in self.model_input_names

						if return_attention_mask and "global_attention_mask" in encoded_inputs:
									_UpperCAmelCase	 =     encoded_inputs[self.model_input_names[0]]
									# `global_attention_mask` need to have the same length as other (sequential) inputs.
									_UpperCAmelCase	 =     len(encoded_inputs['global_attention_mask']) != len(A)

									if needs_to_be_padded:
												_UpperCAmelCase	 =     len(A) - len(encoded_inputs['global_attention_mask'])

												if self.padding_side == "right":
															# Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
															_UpperCAmelCase	 =     (
															    encoded_inputs['global_attention_mask'] + [-1] * difference
															)
												elif self.padding_side == "left":
															_UpperCAmelCase	 =     [-1] * difference + encoded_inputs[
															    'global_attention_mask'
															]
												else:
															raise ValueError('Invalid padding strategy:' + str(self.padding_side))

						return encoded_inputs





import unittest

from transformers import is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
							from PIL import Image
else:







							class     __lowerCAmelCase					:



										@staticmethod
										def 					_lowerCamelCase (		*A     :						Union[str, Any]				,	**A     :						List[Any])     ->					Union[str, Any]:



													"""simple docstring"""






													pass







@is_pipeline_test
@require_vision
class     __lowerCAmelCase					(    unittest.TestCase		):



			@require_torch
			def 					_lowerCamelCase (		self     :						List[str])     ->					Tuple:



						"""simple docstring"""






						_UpperCAmelCase	 =     pipeline(
						    model='hf-internal-testing/tiny-random-clip-zero-shot-image-classification'				,	)
						_UpperCAmelCase	 =     Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png')
						_UpperCAmelCase	 =     image_classifier(A				,	candidate_labels=['a', 'b', 'c'])

						# The floating scores are so close, we enter floating error approximation and the order is not guaranteed across
						# python and torch versions.
						self.assertIn(
						    nested_simplify(A)				,	[
						        [{'score': 0.3_3_3, 'label': 'a'}, {'score': 0.3_3_3, 'label': 'b'}, {'score': 0.3_3_3, 'label': 'c'}],
						        [{'score': 0.3_3_3, 'label': 'a'}, {'score': 0.3_3_3, 'label': 'c'}, {'score': 0.3_3_3, 'label': 'b'}],
						    ]				,	)

						_UpperCAmelCase	 =     image_classifier([image] * 5				,	candidate_labels=['A', 'B', 'C']				,	batch_size=2)
						self.assertEqual(
						    nested_simplify(A)				,	[
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						    ]				,	)



			@require_tf
			def 					_lowerCamelCase (		self     :						str)     ->					Tuple:



						"""simple docstring"""






						_UpperCAmelCase	 =     pipeline(
						    model='hf-internal-testing/tiny-random-clip-zero-shot-image-classification'				,	framework='tf')
						_UpperCAmelCase	 =     Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png')
						_UpperCAmelCase	 =     image_classifier(A				,	candidate_labels=['a', 'b', 'c'])

						self.assertEqual(
						    nested_simplify(A)				,	[{'score': 0.3_3_3, 'label': 'a'}, {'score': 0.3_3_3, 'label': 'b'}, {'score': 0.3_3_3, 'label': 'c'}]				,	)

						_UpperCAmelCase	 =     image_classifier([image] * 5				,	candidate_labels=['A', 'B', 'C']				,	batch_size=2)
						self.assertEqual(
						    nested_simplify(A)				,	[
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						        [
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						            {'score': 0.3_3_3, 'label': ANY(A)},
						        ],
						    ]				,	)



			@slow
			@require_torch
			def 					_lowerCamelCase (		self     :						Tuple)     ->					Optional[Any]:



						"""simple docstring"""






						_UpperCAmelCase	 =     pipeline(
						    task='zero-shot-image-classification'				,	model='openai/clip-vit-base-patch32'				,	)
						# This is an image of 2 cats with remotes and no planes
						_UpperCAmelCase	 =     Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png')
						_UpperCAmelCase	 =     image_classifier(A				,	candidate_labels=['cat', 'plane', 'remote'])

						self.assertEqual(
						    nested_simplify(A)				,	[
						        {'score': 0.5_1_1, 'label': 'remote'},
						        {'score': 0.4_8_5, 'label': 'cat'},
						        {'score': 0.0_0_4, 'label': 'plane'},
						    ]				,	)

						_UpperCAmelCase	 =     image_classifier([image] * 5				,	candidate_labels=['cat', 'plane', 'remote']				,	batch_size=2)
						self.assertEqual(
						    nested_simplify(A)				,	[
						        [
						            {'score': 0.5_1_1, 'label': 'remote'},
						            {'score': 0.4_8_5, 'label': 'cat'},
						            {'score': 0.0_0_4, 'label': 'plane'},
						        ],
						    ]
						    * 5				,	)



			@slow
			@require_tf
			def 					_lowerCamelCase (		self     :						List[str])     ->					Any:



						"""simple docstring"""






						_UpperCAmelCase	 =     pipeline(
						    task='zero-shot-image-classification'				,	model='openai/clip-vit-base-patch32'				,	framework='tf')
						# This is an image of 2 cats with remotes and no planes
						_UpperCAmelCase	 =     Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png')
						_UpperCAmelCase	 =     image_classifier(A				,	candidate_labels=['cat', 'plane', 'remote'])
						self.assertEqual(
						    nested_simplify(A)				,	[
						        {'score': 0.5_1_1, 'label': 'remote'},
						        {'score': 0.4_8_5, 'label': 'cat'},
						        {'score': 0.0_0_4, 'label': 'plane'},
						    ]				,	)

						_UpperCAmelCase	 =     image_classifier([image] * 5				,	candidate_labels=['cat', 'plane', 'remote']				,	batch_size=2)
						self.assertEqual(
						    nested_simplify(A)				,	[
						        [
						            {'score': 0.5_1_1, 'label': 'remote'},
						            {'score': 0.4_8_5, 'label': 'cat'},
						            {'score': 0.0_0_4, 'label': 'plane'},
						        ],
						    ]
						    * 5				,	)
