


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(a)
class      lowerCAmelCase__       (				a):




     '''simple docstring'''


     def __init__(						self ,						*__lowerCamelCase ,						**__lowerCamelCase)			->				List[str]:
            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)			->				Dict:
            _A      :       Optional[int]				    = {}
            _A      :       Tuple				    = {}

            if prompt is not None:
                   _A      :       Dict				    = 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      :       int				    = {}
                   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      :       Dict				    = max_new_tokens
            return preprocess_params, forward_kwargs, {}


     def __call__(						self ,						__lowerCamelCase ,						**__lowerCamelCase)			->				Optional[int]:
            return super().__call__(__lowerCamelCase ,						**__lowerCamelCase)


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase=None)			->				Optional[Any]:
            _A      :       List[Any]				    = 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      :       Optional[Any]				    = self.model.config.model_type

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

                   elif model_type == "pix2struct":
                          _A      :       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
                          _A      :       Optional[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      :       Any				    = self.image_processor(images=__lowerCamelCase ,						return_tensors=self.framework)

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

            return model_inputs


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase=None)			->				Dict:
            # 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      :       Tuple				    = None

            if generate_kwargs is None:
                   _A      :       List[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      :       List[str]				    = model_inputs.pop(self.model.main_input_name)
            _A      :       int				    = self.model.generate(__lowerCamelCase ,						**__lowerCamelCase ,						**__lowerCamelCase)
            return model_outputs


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				Union[str, Any]:
            _A      :       List[str]				    = []
            for output_ids in model_outputs:
                   _A      :       Any				    = {
                       "generated_text": self.tokenizer.decode(
                           __lowerCamelCase ,						skip_special_tokens=__lowerCamelCase ,						)
                   }
                   records.append(__lowerCamelCase)
            return records






from typing import List, Optional

import numpy as np

from ...processing_utils import ProcessorMixin
from ...utils import to_numpy







class     __lowerCAmelCase (       __magic_name__ ):

			"""simple docstring"""







			snake_case_   		=       '''EncodecFeatureExtractor'''
			snake_case_   		=       ('''T5Tokenizer''', '''T5TokenizerFast''')


			def __init__(      self			,						lowerCamelCase__			,						lowerCamelCase__							)    ->      int:



							'''simple docstring'''
							super().__init__(lowerCamelCase__			,						lowerCamelCase__							)
							__lowerCamelCase		       =    self.feature_extractor
							__lowerCamelCase		       =    False


			def       lowercase_    (      self			,						lowerCamelCase__=None			,						lowerCamelCase__=None			,						lowerCamelCase__=True							)    ->      List[Any]:



							'''simple docstring'''
							return self.tokenizer.get_decoder_prompt_ids(task=lowerCamelCase__			,						language=lowerCamelCase__			,						no_timestamps=lowerCamelCase__							)


			def __call__(      self			,						*lowerCamelCase__			,						**lowerCamelCase__							)    ->      Dict:



							'''simple docstring'''
							# For backward compatibility
							if self._in_target_context_manager:
											return self.current_processor(*lowerCamelCase__			,						**lowerCamelCase__							)

							__lowerCamelCase		       =    kwargs.pop('audio'			,						lowerCamelCase__							)
							__lowerCamelCase		       =    kwargs.pop('sampling_rate'			,						lowerCamelCase__							)
							__lowerCamelCase		       =    kwargs.pop('text'			,						lowerCamelCase__							)
							if len(lowerCamelCase__							) > 0:
											__lowerCamelCase		       =    args[0]
											__lowerCamelCase		       =    args[1:]

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

							if text is not None:
											__lowerCamelCase		       =    self.tokenizer(lowerCamelCase__			,						**lowerCamelCase__							)

							if audio is not None:
											__lowerCamelCase		       =    self.feature_extractor(lowerCamelCase__			,						*lowerCamelCase__			,						sampling_rate=lowerCamelCase__			,						**lowerCamelCase__							)

							if audio is None:
											return inputs

							elif text is None:
											return audio_inputs

							else:
											__lowerCamelCase		       =    audio_inputs['input_values']
											if "padding_mask" in audio_inputs:
															__lowerCamelCase		       =    audio_inputs['padding_mask']
											return inputs


			def       lowercase_    (      self			,						*lowerCamelCase__			,						**lowerCamelCase__							)    ->      Union[str, Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    kwargs.pop('audio'			,						lowerCamelCase__							)
							__lowerCamelCase		       =    kwargs.pop('padding_mask'			,						lowerCamelCase__							)

							if len(lowerCamelCase__							) > 0:
											__lowerCamelCase		       =    args[0]
											__lowerCamelCase		       =    args[1:]

							if audio_values is not None:
											return self._decode_audio(lowerCamelCase__			,						padding_mask=lowerCamelCase__							)
							else:
											return self.tokenizer.batch_decode(*lowerCamelCase__			,						**lowerCamelCase__							)


			def       lowercase_    (      self			,						*lowerCamelCase__			,						**lowerCamelCase__							)    ->      List[Any]:



							'''simple docstring'''
							return self.tokenizer.decode(*lowerCamelCase__			,						**lowerCamelCase__							)






			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = None							)    ->      List[np.ndarray]:



							'''simple docstring'''
							__lowerCamelCase		       =    to_numpy(lowerCamelCase__							)
							__lowerCamelCase       ,						__lowerCamelCase       ,						__lowerCamelCase		       =    audio_values.shape

							if padding_mask is None:
											return list(lowerCamelCase__							)

							__lowerCamelCase		       =    to_numpy(lowerCamelCase__							)

							# match the sequence length of the padding mask to the generated audio arrays by padding with the **non-padding**
							# token (so that the generated audio values are **not** treated as padded tokens)
							__lowerCamelCase		       =    seq_len - padding_mask.shape[-1]
							__lowerCamelCase		       =    1 - self.feature_extractor.padding_value
							__lowerCamelCase		       =    np.pad(lowerCamelCase__			,						((0, 0), (0, difference))			,						'constant'			,						constant_values=lowerCamelCase__							)

							__lowerCamelCase		       =    audio_values.tolist()
							for i in range(lowerCamelCase__							):
											__lowerCamelCase		       =    np.asarray(audio_values[i]							)[
											    padding_mask[i][None, :] != self.feature_extractor.padding_value
											]
											__lowerCamelCase		       =    sliced_audio.reshape(lowerCamelCase__			,						-1							)

							return audio_values


