

"""simple docstring"""




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   lowerCAmelCase						(					):
	'''simple docstring'''





	UpperCAmelCase__					:  Any					   =					(
	    list(range(ord("""!"""      )							,    ord("""~"""      ) + 1      )      ) + list(range(ord("""¡"""      )							,    ord("""¬"""      ) + 1      )      ) + list(range(ord("""®"""      )							,    ord("""ÿ"""      ) + 1      )      )
	)
	UpperCAmelCase__					:  Union[str, Any]					   =					bs[:]
	UpperCAmelCase__					:  Union[str, Any]					   =					0
	for b in range(2**8      ):
		if b not in bs:
			bs.append(_lowerCamelCase      )
			cs.append(2**8 + n      )
			n += 1
	UpperCAmelCase__					:  str					   =					[chr(_lowerCamelCase      ) for n in cs]
	return dict(zip(_lowerCamelCase							,    _lowerCamelCase      )      )
def   lowerCAmelCase						(					__UpperCamelCase      ):
	'''simple docstring'''





	UpperCAmelCase__					:  int					   =					set()
	UpperCAmelCase__					:  Optional[Any]					   =					word[0]
	for char in word[1:]:
		pairs.add((prev_char, char)      )
		UpperCAmelCase__					:  Optional[Any]					   =					char
	return pairs







class      __lowercase			(    a__					):
		snake_case_				       =       VOCAB_FILES_NAMES
		snake_case_				       =       PRETRAINED_VOCAB_FILES_MAP
		snake_case_				       =       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		snake_case_				       =       ["""input_ids""", """attention_mask"""]

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






			'''simple docstring'''
			UpperCAmelCase__					:  Optional[int]					   =					AddedToken(_A					,lstrip=_A					,rstrip=_A    ) if isinstance(_A					,_A    ) else bos_token
			UpperCAmelCase__					:  Dict					   =					AddedToken(_A					,lstrip=_A					,rstrip=_A    ) if isinstance(_A					,_A    ) else eos_token
			UpperCAmelCase__					:  Union[str, Any]					   =					AddedToken(_A					,lstrip=_A					,rstrip=_A    ) if isinstance(_A					,_A    ) else sep_token
			UpperCAmelCase__					:  Any					   =					AddedToken(_A					,lstrip=_A					,rstrip=_A    ) if isinstance(_A					,_A    ) else cls_token
			UpperCAmelCase__					:  Tuple					   =					AddedToken(_A					,lstrip=_A					,rstrip=_A    ) if isinstance(_A					,_A    ) else unk_token
			UpperCAmelCase__					:  Optional[int]					   =					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__					:  str					   =					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__					:  Tuple					   =					json.load(_A    )
			UpperCAmelCase__					:  Optional[int]					   =					{v: k for k, v in self.encoder.items()}
			UpperCAmelCase__					:  Optional[Any]					   =					errors  # how to handle errors in decoding
			UpperCAmelCase__					:  Optional[int]					   =					bytes_to_unicode()
			UpperCAmelCase__					:  int					   =					{v: k for k, v in self.byte_encoder.items()}
			with open(_A					,encoding="""utf-8"""    ) as merges_handle:
				UpperCAmelCase__					:  Tuple					   =					merges_handle.read().split("""\n"""    )[1:-1]
			UpperCAmelCase__					:  Optional[Any]					   =					[tuple(merge.split()    ) for merge in bpe_merges]
			UpperCAmelCase__					:  Optional[int]					   =					dict(zip(_A					,range(len(_A    )    )    )    )
			UpperCAmelCase__					:  Any					   =					{}
			UpperCAmelCase__					:  List[Any]					   =					add_prefix_space

			# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
			UpperCAmelCase__					:  int					   =					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    __lowercase       (  self		:			Union[str, Any]    ):






			'''simple docstring'''
			return len(self.encoder    )

		def    __lowercase       (  self		:			Any    ):






			'''simple docstring'''
			return dict(self.encoder					,**self.added_tokens_encoder    )

		def    __lowercase       (  self		:			Tuple					,A		:			Dict    ):






			'''simple docstring'''

			if token in self.cache:
				return self.cache[token]
			UpperCAmelCase__					:  int					   =					tuple(_A    )
			UpperCAmelCase__					:  str					   =					get_pairs(_A    )

			if not pairs:
				return token

			while True:
				UpperCAmelCase__					:  Dict					   =					min(_A					,key=lambda A    : self.bpe_ranks.get(_A					,float("""inf"""    )    )    )
				if bigram not in self.bpe_ranks:
					break
				UpperCAmelCase__					:  List[str]					   =					bigram
				UpperCAmelCase__					:  Tuple					   =					[]
				UpperCAmelCase__					:  Tuple					   =					0
				while i < len(_A    ):
					try:
						UpperCAmelCase__					:  int					   =					word.index(_A					,_A    )
					except ValueError:
						new_word.extend(word[i:]    )
						break
					else:
						new_word.extend(word[i:j]    )
						UpperCAmelCase__					:  int					   =					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__					:  Optional[Any]					   =					tuple(_A    )
				UpperCAmelCase__					:  Tuple					   =					new_word
				if len(_A    ) == 1:
					break
				else:
					UpperCAmelCase__					:  List[str]					   =					get_pairs(_A    )
			UpperCAmelCase__					:  Tuple					   =					' '.join(_A    )
			UpperCAmelCase__					:  List[Any]					   =					word
			return word

		def    __lowercase       (  self		:			str					,A		:			Union[str, Any]    ):






			'''simple docstring'''
			UpperCAmelCase__					:  Optional[int]					   =					[]
			for token in re.findall(self.pat					,_A    ):
				UpperCAmelCase__					:  int					   =					''.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    __lowercase       (  self		:			Any					,A		:			Dict    ):






			'''simple docstring'''
			return self.encoder.get(_A					,self.encoder.get(self.unk_token    )    )

		def    __lowercase       (  self		:			Union[str, Any]					,A		:			Any    ):






			'''simple docstring'''
			return self.decoder.get(_A    )

		def    __lowercase       (  self		:			List[str]					,A		:			int    ):






			'''simple docstring'''
			UpperCAmelCase__					:  int					   =					''.join(_A    )
			UpperCAmelCase__					:  Optional[Any]					   =					bytearray([self.byte_decoder[c] for c in text]    ).decode("""utf-8"""					,errors=self.errors    )
			return text

		def    __lowercase       (  self		:			Tuple					,A		:			str					,A		:			List[str] = None    ):






			'''simple docstring'''

			if not os.path.isdir(_A    ):
				logger.error(f"Vocabulary path ({save_directory}) should be a directory"    )
				return
			UpperCAmelCase__					:  Any					   =					os.path.join(
			    _A					,(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]    )
			UpperCAmelCase__					:  List[str]					   =					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__					:  Optional[int]					   =					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__					:  Union[str, Any]					   =					token_index
					writer.write(""" """.join(_A    ) + """\n"""    )
					index += 1

			return vocab_file, merge_file

		def    __lowercase       (  self		:			List[Any]					,A		:			Optional[int]					,A		:			str = None    ):






			'''simple docstring'''

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

		def    __lowercase       (  self		:			Optional[int]					,A		:			Dict					,A		:			int = None					,A		:			Optional[int] = False    ):






			'''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    __lowercase       (  self		:			List[str]					,A		:			Dict					,A		:			Optional[int] = None    ):






			'''simple docstring'''
			UpperCAmelCase__					:  int					   =					[self.sep_token_id]
			UpperCAmelCase__					:  str					   =					[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    __lowercase       (  self		:			Optional[Any]					,A		:			str					,A		:			str=False					,**A		:			int    ):






			'''simple docstring'''
			UpperCAmelCase__					:  List[str]					   =					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__					:  Union[str, Any]					   =					' ' + text
			return (text, kwargs)




		def    __lowercase       (  self		:			List[str]					,A		:			Optional[Any]					,A		:			str = None					,A		:			str = PaddingStrategy.DO_NOT_PAD					,A		:			Optional[Any] = None					,A		:			List[Any] = None					,):






			'''simple docstring'''
			UpperCAmelCase__					:  Any					   =					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__					:  Union[str, Any]					   =					'attention_mask' in self.model_input_names

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

				if needs_to_be_padded:
					UpperCAmelCase__					:  Optional[Any]					   =					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__					:  Optional[Any]					   =					(
						    encoded_inputs['global_attention_mask'] + [-1] * difference
						)
					elif self.padding_side == "left":
						UpperCAmelCase__					:  List[Any]					   =					[-1] * difference + encoded_inputs[
						    'global_attention_mask'
						]
					else:
						raise ValueError("""Invalid padding strategy:""" + str(self.padding_side    )    )

			return encoded_inputs








"""simple docstring"""




from typing import Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
      from PIL import Image

      from ..image_utils import load_image

if is_torch_available():
      from ..models.auto.modeling_auto import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING

__UpperCAmelCase =				logging.get_logger(__name__)







@add_end_docstrings(__lowerCamelCase					)
class      __lowercase			(    __lowerCamelCase					):

  def __init__(  self		:			List[str]					,*A		:			Optional[int]					,**A		:			List[str]    ):






   '''simple docstring'''
   super().__init__(*A					,**A    )
   self.check_model_type(A    )

  def    __lowercase       (  self		:			Optional[int]					,A		:			Dict=None					,A		:			Optional[int]=None					,A		:			Tuple=None					,**A		:			str    ):






   '''simple docstring'''
   UpperCAmelCase__   ,      UpperCAmelCase__					:  Any					   =					{}, {}
   if padding is not None:
    UpperCAmelCase__					:  Union[str, Any]					   =					padding
   if truncation is not None:
    UpperCAmelCase__					:  List[str]					   =					truncation
   if top_k is not None:
    UpperCAmelCase__					:  List[Any]					   =					top_k
   return preprocess_params, {}, postprocess_params

  def __call__(  self		:			List[Any]					,A		:			Union["Image.Image", str]					,A		:			str = None					,**A		:			Union[str, Any]    ):






   '''simple docstring'''
   if isinstance(A					,(Image.Image, str)    ) and isinstance(A					,A    ):
    UpperCAmelCase__					:  Any					   =					{"""image""": image, """question""": question}
   else:
    UpperCAmelCase__					:  Dict					   =					image
   UpperCAmelCase__					:  Dict					   =					super().__call__(A					,**A    )
   return results

  def    __lowercase       (  self		:			Optional[int]					,A		:			Optional[Any]					,A		:			Optional[Any]=False					,A		:			List[Any]=False    ):






   '''simple docstring'''
   UpperCAmelCase__					:  List[Any]					   =					load_image(inputs["""image"""]    )
   UpperCAmelCase__					:  Optional[Any]					   =					self.tokenizer(
       inputs["""question"""]					,return_tensors=self.framework					,padding=A					,truncation=A    )
   UpperCAmelCase__					:  int					   =					self.image_processor(images=A					,return_tensors=self.framework    )
   model_inputs.update(A    )
   return model_inputs

  def    __lowercase       (  self		:			Dict					,A		:			Tuple    ):






   '''simple docstring'''
   UpperCAmelCase__					:  List[Any]					   =					self.model(**A    )
   return model_outputs




  def    __lowercase       (  self		:			int					,A		:			List[Any]					,A		:			List[str]=5    ):






   '''simple docstring'''
   if top_k > self.model.config.num_labels:
    UpperCAmelCase__					:  Any					   =					self.model.config.num_labels

   if self.framework == "pt":
    UpperCAmelCase__					:  Optional[int]					   =					model_outputs.logits.sigmoid()[0]
    UpperCAmelCase__   ,      UpperCAmelCase__					:  Optional[int]					   =					probs.topk(A    )
   else:
    raise ValueError(f"Unsupported framework: {self.framework}"    )

   UpperCAmelCase__					:  List[str]					   =					scores.tolist()
   UpperCAmelCase__					:  Optional[int]					   =					ids.tolist()
   return [{"score": score, "answer": self.model.config.idalabel[_id]} for score, _id in zip(A					,A    )]





