






"""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

_lowerCAmelCase         =							logging.get_logger(__name__)



@add_end_docstrings(_SCREAMING_SNAKE_CASE						)
class      SCREAMING_SNAKE_CASE  (				_SCREAMING_SNAKE_CASE						):





	"""simple docstring"""




	def __init__(self     ,     *lowerCAmelCase_     ,     **lowerCAmelCase_							):
		super().__init__(*lowerCAmelCase_     ,     **lowerCAmelCase_							)
		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    lowerCamelCase(self     ,     lowerCAmelCase_=None     ,     lowerCAmelCase_=None     ,     lowerCAmelCase_=None							):
		A_		:						Optional[Any]	    =   {}
		A_		:						Any	    =   {}

		if prompt is not None:
			A_		:						List[str]	    =   prompt

		if generate_kwargs is not None:
			A_		:						Union[str, Any]	    =   generate_kwargs
		if max_new_tokens is not None:
			if "generate_kwargs" not in forward_kwargs:
				A_		:						List[str]	    =   {}
			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"""							)
			A_		:						List[str]	    =   max_new_tokens
		return preprocess_params, forward_kwargs, {}





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





	def    lowerCamelCase(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_=None							):
		A_		:						Optional[int]	    =   load_image(lowerCAmelCase_							)

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

			A_		:						List[Any]	    =   self.model.config.model_type

			if model_type == "git":
				A_		:						Optional[int]	    =   self.image_processor(images=lowerCAmelCase_     ,     return_tensors=self.framework							)
				A_		:						Tuple	    =   self.tokenizer(text=lowerCAmelCase_     ,     add_special_tokens=lowerCAmelCase_							).input_ids
				A_		:						str	    =   [self.tokenizer.cls_token_id] + input_ids
				A_		:						List[Any]	    =   torch.tensor(lowerCAmelCase_							).unsqueeze(0							)
				model_inputs.update({"""input_ids""": input_ids}							)

			elif model_type == "pix2struct":
				A_		:						List[Any]	    =   self.image_processor(images=lowerCAmelCase_     ,     header_text=lowerCAmelCase_     ,     return_tensors=self.framework							)

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

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

		else:
			A_		:						Optional[int]	    =   self.image_processor(images=lowerCAmelCase_     ,     return_tensors=self.framework							)

		if self.model.config.model_type == "git" and prompt is None:
			A_		:						Optional[int]	    =   None

		return model_inputs





	def    lowerCamelCase(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_=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"""]     ,     lowerCAmelCase_							)
		    and all(x is None for x in model_inputs["""input_ids"""]							)
		):
			A_		:						List[Any]	    =   None

		if generate_kwargs is None:
			A_		:						Union[str, Any]	    =   {}
		# 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.
		A_		:						Any	    =   model_inputs.pop(self.model.main_input_name							)
		A_		:						int	    =   self.model.generate(lowerCAmelCase_     ,     **lowerCAmelCase_     ,     **lowerCAmelCase_							)
		return model_outputs





	def    lowerCamelCase(self     ,     lowerCAmelCase_							):
		A_		:						Dict	    =   []
		for output_ids in model_outputs:
			A_		:						Union[str, Any]	    =   {
			    """generated_text""": self.tokenizer.decode(
			        lowerCAmelCase_     ,     skip_special_tokens=lowerCAmelCase_     ,     )
			}
			records.append(lowerCAmelCase_							)
		return records









"""simple docstring"""
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding



class      SCREAMING_SNAKE_CASE  (				_SCREAMING_SNAKE_CASE						):





	"""simple docstring"""




	_A       :				Optional[int]    =				["""image_processor""", """tokenizer"""]
	_A       :				str    =				"""AutoImageProcessor"""
	_A       :				Union[str, Any]    =				"""AutoTokenizer"""





	def __init__(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_							):
		super().__init__(lowerCAmelCase_     ,     lowerCAmelCase_							)
		A_		:						Union[str, Any]	    =   self.image_processor





	def __call__(self     ,     lowerCAmelCase_=None     ,     lowerCAmelCase_=None     ,     lowerCAmelCase_=None     ,     **lowerCAmelCase_							):

		if text is None and images is None:
			raise ValueError("""You have to specify either text or images. Both cannot be none."""							)

		if text is not None:
			A_		:						Optional[int]	    =   self.tokenizer(lowerCAmelCase_     ,     return_tensors=lowerCAmelCase_     ,     **lowerCAmelCase_							)

		if images is not None:
			A_		:						List[Any]	    =   self.image_processor(lowerCAmelCase_     ,     return_tensors=lowerCAmelCase_     ,     **lowerCAmelCase_							)

		if text is not None and images is not None:
			A_		:						int	    =   image_features.pixel_values
			return encoding
		elif text is not None:
			return encoding
		else:
			return BatchEncoding(data=dict(**lowerCAmelCase_							)     ,     tensor_type=lowerCAmelCase_							)





	def    lowerCamelCase(self     ,     *lowerCAmelCase_     ,     **lowerCAmelCase_							):
		return self.tokenizer.batch_decode(*lowerCAmelCase_     ,     **lowerCAmelCase_							)





	def    lowerCamelCase(self     ,     *lowerCAmelCase_     ,     **lowerCAmelCase_							):
		return self.tokenizer.decode(*lowerCAmelCase_     ,     **lowerCAmelCase_							)





	@property
	def    lowerCamelCase(self							):
		return ["input_ids", "attention_mask", "pixel_values"]

