


"""simple docstring"""


from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
	from PIL import Image

	from ..image_utils import load_image

if is_tf_available():
	from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_VISION_2_SEQ_MAPPING

if is_torch_available():
	import torch

	from ..models.auto.modeling_auto import MODEL_FOR_VISION_2_SEQ_MAPPING

A     :   int			    =			logging.get_logger(__name__)
@add_end_docstrings(lowerCAmelCase__		)
class   _UpperCamelCase  (   lowerCAmelCase__		):
							'''simple docstring'''







							def __init__(     self    ,   *__a    ,   **__a	):
										super().__init__(*__a    ,   **__a	)
										requires_backends(self    ,   "vision"	)
										self.check_model_type(
										    TF_MODEL_FOR_VISION_2_SEQ_MAPPING if self.framework == "tf" else MODEL_FOR_VISION_2_SEQ_MAPPING	)

							def 			snake_case				(     self    ,   __a=None    ,   __a=None    ,   __a=None	):
										__lowerCAmelCase				    =			{}
										__lowerCAmelCase				    =			{}

										if prompt is not None:
													__lowerCAmelCase				    =			prompt

										if generate_kwargs is not None:
													__lowerCAmelCase				    =			generate_kwargs
										if max_new_tokens is not None:
													if "generate_kwargs" not in forward_kwargs:
																__lowerCAmelCase				    =			{}
													if "max_new_tokens" in forward_kwargs["generate_kwargs"]:
																raise ValueError(
																    "'max_new_tokens' is defined twice, once in 'generate_kwargs' and once as a direct parameter,"
																    " please use only one"	)
													__lowerCAmelCase				    =			max_new_tokens
										return preprocess_params, forward_kwargs, {}

							def __call__(     self    ,   __a    ,   **__a	):
										return super().__call__(__a    ,   **__a	)

							def 			snake_case				(     self    ,   __a    ,   __a=None	):
										__lowerCAmelCase				    =			load_image(__a	)

										if prompt is not None:
													if not isinstance(__a    ,   __a	):
																raise ValueError(
																    f"Received an invalid text input, got - {type(__a	)} - but expected a single string. "
																    "Note also that one single text can be provided for conditional image to text generation."	)

													__lowerCAmelCase				    =			self.model.config.model_type

													if model_type == "git":
																__lowerCAmelCase				    =			self.image_processor(images=__a    ,   return_tensors=self.framework	)
																__lowerCAmelCase				    =			self.tokenizer(text=__a    ,   add_special_tokens=__a	).input_ids
																__lowerCAmelCase				    =			[self.tokenizer.cls_token_id] + input_ids
																__lowerCAmelCase				    =			torch.tensor(__a	).unsqueeze(0	)
																model_inputs.update({"input_ids": input_ids}	)

													elif model_type == "pix2struct":
																__lowerCAmelCase				    =			self.image_processor(images=__a    ,   header_text=__a    ,   return_tensors=self.framework	)

													elif model_type != "vision-encoder-decoder":
																# vision-encoder-decoder does not support conditional generation
																__lowerCAmelCase				    =			self.image_processor(images=__a    ,   return_tensors=self.framework	)
																__lowerCAmelCase				    =			self.tokenizer(__a    ,   return_tensors=self.framework	)
																model_inputs.update(__a	)

													else:
																raise ValueError(f"Model type {model_type} does not support conditional text generation"	)

										else:
													__lowerCAmelCase				    =			self.image_processor(images=__a    ,   return_tensors=self.framework	)

										if self.model.config.model_type == "git" and prompt is None:
													__lowerCAmelCase				    =			None

										return model_inputs

							def 			snake_case				(     self    ,   __a    ,   __a=None	):
										# Git model sets `model_inputs["input_ids"] = None` in `preprocess` (when `prompt=None`). In batch model, the
										# pipeline will group them into a list of `None`, which fail `_forward`. Avoid this by checking it first.
										if (
										    "input_ids" in model_inputs
										    and isinstance(model_inputs["input_ids"]    ,   __a	)
										    and all(x is None for x in model_inputs["input_ids"]	)
										):
													__lowerCAmelCase				    =			None

										if generate_kwargs is None:
													__lowerCAmelCase				    =			{}
										# FIXME: We need to pop here due to a difference in how `generation.py` and `generation.tf_utils.py`
										#  parse inputs. In the Tensorflow version, `generate` raises an error if we don't use `input_ids` whereas
										#  the PyTorch version matches it with `self.model.main_input_name` or `self.model.encoder.main_input_name`
										#  in the `_prepare_model_inputs` method.
										__lowerCAmelCase				    =			model_inputs.pop(self.model.main_input_name	)
										__lowerCAmelCase				    =			self.model.generate(__a    ,   **__a    ,   **__a	)
										return model_outputs



							def 			snake_case				(     self    ,   __a	):
										__lowerCAmelCase				    =			[]
										for output_ids in model_outputs:
													__lowerCAmelCase				    =			{
													    "generated_text": self.tokenizer.decode(
													        __a    ,   skip_special_tokens=__a    ,   )
													}
													records.append(__a	)
										return records








"""simple docstring"""


import warnings
from typing import Dict, List, Optional, Tuple

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


A     :   str			    =			logging.get_logger(__name__)
class   _UpperCamelCase  (   lowerCAmelCase__		):
							'''simple docstring'''







							__UpperCAmelCase						:    str				      =["""input_ids""", """attention_mask"""]

							def __init__(     self    ,   __a="</s>"    ,   __a="<unk>"    ,   __a="<pad>"    ,   __a=1_25    ,   __a=None    ,   **__a    ,   ):
										# Add extra_ids to the special token list
										if extra_ids > 0 and additional_special_tokens is None:
													__lowerCAmelCase				    =			[f"<extra_id_{i}>" for i in range(__a	)]
										elif extra_ids > 0 and additional_special_tokens is not None:
													# Check that we have the right number of extra_id special tokens
													__lowerCAmelCase				    =			len(set(filter(lambda __a	: bool("extra_id" in str(__a	)	)    ,   __a	)	)	)
													if extra_tokens != extra_ids:
																raise ValueError(
																    f"Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are"
																    " provided to ByT5Tokenizer. In this case the additional_special_tokens must include the"
																    " extra_ids tokens"	)

										__lowerCAmelCase				    =			AddedToken(__a    ,   lstrip=__a    ,   rstrip=__a	) if isinstance(__a    ,   __a	) else pad_token
										__lowerCAmelCase				    =			AddedToken(__a    ,   lstrip=__a    ,   rstrip=__a	) if isinstance(__a    ,   __a	) else eos_token
										__lowerCAmelCase				    =			AddedToken(__a    ,   lstrip=__a    ,   rstrip=__a	) if isinstance(__a    ,   __a	) else unk_token

										super().__init__(
										    eos_token=__a    ,   unk_token=__a    ,   pad_token=__a    ,   extra_ids=__a    ,   additional_special_tokens=__a    ,   **__a    ,   )

										__lowerCAmelCase				    =			extra_ids

										__lowerCAmelCase				    =			2**8  # utf is 8 bits

										# define special tokens dict
										__lowerCAmelCase				    =			{
										    self.pad_token: 0,
										    self.eos_token: 1,
										    self.unk_token: 2,
										}
										__lowerCAmelCase				    =			len(self.special_tokens_encoder	)
										__lowerCAmelCase				    =			len(__a	)
										for i, token in enumerate(__a	):
													__lowerCAmelCase				    =			self.vocab_size + i - n
										__lowerCAmelCase				    =			{v: k for k, v in self.special_tokens_encoder.items()}

							@property
							def 			snake_case				(     self	):
										return self._utf_vocab_size + self._num_special_tokens + self._extra_ids

							def 			snake_case				(     self    ,   __a    ,   __a = None    ,   __a = False	):
										if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=__a    ,   token_ids_a=__a    ,   already_has_special_tokens=__a	)

										# normal case: some special tokens
										if token_ids_a is None:
													return ([0] * len(__a	)) + [1]
										return ([0] * len(__a	)) + [1] + ([0] * len(__a	)) + [1]

							def 			snake_case				(     self    ,   __a	):
										if len(__a	) > 0 and token_ids[-1] == self.eos_token_id:
													warnings.warn(
													    f"This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated"
													    " eos tokens being added."	)
													return token_ids
										else:
													return token_ids + [self.eos_token_id]

							def 			snake_case				(     self    ,   __a    ,   __a = None	):
										__lowerCAmelCase				    =			[self.eos_token_id]

										if token_ids_a is None:
													return len(token_ids_a + eos	) * [0]
										return len(token_ids_a + eos + token_ids_a + eos	) * [0]

							def 			snake_case				(     self    ,   __a    ,   __a = None	):
										__lowerCAmelCase				    =			self._add_eos_if_not_present(__a	)
										if token_ids_a is None:
													return token_ids_a
										else:
													__lowerCAmelCase				    =			self._add_eos_if_not_present(__a	)
													return token_ids_a + token_ids_a

							def 			snake_case				(     self    ,   __a	):
										__lowerCAmelCase				    =			[chr(__a	) for i in text.encode("utf-8"	)]
										return tokens

							def 			snake_case				(     self    ,   __a	):
										if token in self.special_tokens_encoder:
													__lowerCAmelCase				    =			self.special_tokens_encoder[token]
										elif token in self.added_tokens_encoder:
													__lowerCAmelCase				    =			self.added_tokens_encoder[token]
										elif len(__a	) != 1:
													__lowerCAmelCase				    =			self.unk_token_id
										else:
													__lowerCAmelCase				    =			ord(__a	) + self._num_special_tokens
										return token_id

							def 			snake_case				(     self    ,   __a	):
										if index in self.special_tokens_decoder:
													__lowerCAmelCase				    =			self.special_tokens_decoder[index]
										else:
													__lowerCAmelCase				    =			chr(index - self._num_special_tokens	)
										return token

							def 			snake_case				(     self    ,   __a	):
										__lowerCAmelCase				    =			B""
										for token in tokens:
													if token in self.special_tokens_decoder:
																__lowerCAmelCase				    =			self.special_tokens_decoder[token].encode("utf-8"	)
													elif token in self.added_tokens_decoder:
																__lowerCAmelCase				    =			self.special_tokens_decoder[token].encode("utf-8"	)
													elif token in self.special_tokens_encoder:
																__lowerCAmelCase				    =			token.encode("utf-8"	)
													elif token in self.added_tokens_encoder:
																__lowerCAmelCase				    =			token.encode("utf-8"	)
													else:
																__lowerCAmelCase				    =			bytes([ord(__a	)]	)
													bstring += tok_string
										__lowerCAmelCase				    =			bstring.decode("utf-8"    ,   errors="ignore"	)
										return string



							def 			snake_case				(     self    ,   __a    ,   __a = None	):
										return ()




