





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

UpperCamelCase_									=logging.get_logger(__name__)




@add_end_docstrings(_lowerCAmelCase    )
class 			_a						(				_lowerCAmelCase    ):






				def __init__(			self		:   int,			*lowerCAmelCase__		:   int,			**lowerCAmelCase__		:   Union[str, Any]							)     ->				Union[str, Any]:




										'''simple docstring'''



										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     snake_case     (			self		:   List[Any],			lowerCAmelCase__		:   Tuple=None,			lowerCAmelCase__		:   int=None,			lowerCAmelCase__		:   Optional[Any]=None							)     ->				List[Any]:




										'''simple docstring'''



										_UpperCamelCase      : Tuple         =   {}
										_UpperCamelCase      : Optional[int]         =   {}

										if prompt is not None:
																_UpperCamelCase      : int         =   prompt

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






				def __call__(			self		:   int,			lowerCAmelCase__		:   Union[str, List[str], "Image.Image", List["Image.Image"]],			**lowerCAmelCase__		:   Dict							)     ->				Tuple:




										'''simple docstring'''



										return super().__call__(lowerCAmelCase__,			**lowerCAmelCase__							)






				def     snake_case     (			self		:   str,			lowerCAmelCase__		:   List[Any],			lowerCAmelCase__		:   Optional[Any]=None							)     ->				List[str]:




										'''simple docstring'''



										_UpperCamelCase      : str         =   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.'''							)

																_UpperCamelCase      : Any         =   self.model.config.model_type

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

																elif model_type == "pix2struct":
																						_UpperCamelCase      : List[str]         =   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
																						_UpperCamelCase      : Optional[int]         =   self.image_processor(images=lowerCAmelCase__,			return_tensors=self.framework							)
																						_UpperCamelCase      : str         =   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:
																_UpperCamelCase      : Optional[int]         =   self.image_processor(images=lowerCAmelCase__,			return_tensors=self.framework							)

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

										return model_inputs






				def     snake_case     (			self		:   Union[str, Any],			lowerCAmelCase__		:   List[str],			lowerCAmelCase__		:   Tuple=None							)     ->				Union[str, Any]:




										'''simple docstring'''



										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''']							)
										):
																_UpperCamelCase      : Optional[Any]         =   None

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






				def     snake_case     (			self		:   Optional[Any],			lowerCAmelCase__		:   Optional[int]							)     ->				Optional[Any]:




										'''simple docstring'''



										_UpperCamelCase      : List[str]         =   []
										for output_ids in model_outputs:
																_UpperCamelCase      : Optional[Any]         =   {
																    '''generated_text''': self.tokenizer.decode(
																        lowerCAmelCase__,			skip_special_tokens=lowerCAmelCase__,			)
																}
																records.append(lowerCAmelCase__							)
										return records










"""simple docstring"""




from ...processing_utils import ProcessorMixin




class 			_a						(				_lowerCAmelCase    ):
				UpperCamelCase								=   ['''image_processor''', '''feature_extractor''']
				UpperCamelCase								=   '''TvltImageProcessor'''
				UpperCamelCase								=   '''TvltFeatureExtractor'''






				def __init__(			self		:   Union[str, Any],			lowerCAmelCase__		:   str,			lowerCAmelCase__		:   List[str]							)     ->				Optional[Any]:




										'''simple docstring'''



										super().__init__(image_processor=lowerCAmelCase__,			feature_extractor=lowerCAmelCase__							)

										_UpperCamelCase      : List[str]         =   image_processor
										_UpperCamelCase      : int         =   feature_extractor






				def __call__(			self		:   List[str],			lowerCAmelCase__		:   Optional[int]=None,			lowerCAmelCase__		:   str=None,			lowerCAmelCase__		:   Dict=None,			lowerCAmelCase__		:   str=None,			lowerCAmelCase__		:   Optional[int]=False,			lowerCAmelCase__		:   str=False,			*lowerCAmelCase__		:   List[str],			**lowerCAmelCase__		:   Optional[int],			)     ->				Dict:




										'''simple docstring'''



										if images is None and audio is None:
																raise ValueError('''You need to specify either an `images` or `audio` input to process.'''							)

										_UpperCamelCase      : Optional[int]         =   None
										if images is not None:
																_UpperCamelCase      : Optional[int]         =   self.image_processor(lowerCAmelCase__,			mask_pixel=lowerCAmelCase__,			*lowerCAmelCase__,			**lowerCAmelCase__							)
										if images_mixed is not None:
																_UpperCamelCase      : str         =   self.image_processor(lowerCAmelCase__,			is_mixed=lowerCAmelCase__,			*lowerCAmelCase__,			**lowerCAmelCase__							)
										if audio is not None:
																_UpperCamelCase      : Union[str, Any]         =   self.feature_extractor(
																    lowerCAmelCase__,			*lowerCAmelCase__,			sampling_rate=lowerCAmelCase__,			mask_audio=lowerCAmelCase__,			**lowerCAmelCase__							)

										_UpperCamelCase      : str         =   {}
										if audio is not None:
																output_dict.update(lowerCAmelCase__							)
										if images is not None:
																output_dict.update(lowerCAmelCase__							)
										if images_mixed_dict is not None:
																output_dict.update(lowerCAmelCase__							)
										return output_dict






				@property
				def     snake_case     (			self		:   List[str]							)     ->				Tuple:




										'''simple docstring'''



										_UpperCamelCase      : List[str]         =   self.image_processor.model_input_names
										_UpperCamelCase      : List[Any]         =   self.feature_extractor.model_input_names
										return list(dict.fromkeys(image_processor_input_names + feature_extractor_input_names							)							)



