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



      def __init__(   self     ,	*UpperCamelCase_     ,	**UpperCamelCase_      ):
          super().__init__(*UpperCamelCase_     ,	**UpperCamelCase_      )
          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     ,	UpperCamelCase_=None     ,	UpperCamelCase_=None     ,	UpperCamelCase_=None      ):
          UpperCAmelCase__	:      str     							=					{}
          UpperCAmelCase__	:      Optional[int]     							=					{}

          if prompt is not None:
              UpperCAmelCase__	:      List[Any]     							=					prompt

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



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



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_=None      ):
          UpperCAmelCase__	:      List[str]     							=					load_image(UpperCamelCase_      )

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

              UpperCAmelCase__	:      int     							=					self.model.config.model_type

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

              elif model_type == "pix2struct":
                  UpperCAmelCase__	:      Optional[int]     							=					self.image_processor(images=UpperCamelCase_     ,	header_text=UpperCamelCase_     ,	return_tensors=self.framework      )

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

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

          else:
              UpperCAmelCase__	:      Tuple     							=					self.image_processor(images=UpperCamelCase_     ,	return_tensors=self.framework      )

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

          return model_inputs



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_=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']     ,	UpperCamelCase_      )
              and all(x is None for x in model_inputs['input_ids']      )
          ):
              UpperCAmelCase__	:      List[str]     							=					None

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



      def  __snake_case							(   self     ,	UpperCamelCase_      ):
          UpperCAmelCase__	:      Tuple     							=					[]
          for output_ids in model_outputs:
              UpperCAmelCase__	:      int     							=					{
                  'generated_text': self.tokenizer.decode(
                      UpperCamelCase_     ,	skip_special_tokens=UpperCamelCase_     ,	)
              }
              records.append(UpperCamelCase_      )
          return records



"""simple docstring"""



import random
import unittest

import torch

from diffusers import IFImgaImgSuperResolutionPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin





@skip_mps
class 				a			(					lowercase						,		lowercase						,		unittest.TestCase      ):
      UpperCamelCase							:   int           =     IFImgaImgSuperResolutionPipeline
      UpperCamelCase							:   Union[str, Any]           =     TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""width""", """height"""}
      UpperCamelCase							:   Tuple           =     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""original_image"""}      )
      UpperCamelCase							:   Union[str, Any]           =     PipelineTesterMixin.required_optional_params - {"""latents"""}



      def  __snake_case							(   self      ):
          return self._get_superresolution_dummy_components()



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_=0      ):
          if str(UpperCamelCase_      ).startswith('mps'      ):
              UpperCAmelCase__	:      Optional[int]     							=					torch.manual_seed(UpperCamelCase_      )
          else:
              UpperCAmelCase__	:      int     							=					torch.Generator(device=UpperCamelCase_      ).manual_seed(UpperCamelCase_      )

          UpperCAmelCase__	:      Union[str, Any]     							=					floats_tensor((1, 3, 32, 32)     ,	rng=random.Random(UpperCamelCase_      )      ).to(UpperCamelCase_      )
          UpperCAmelCase__	:      Tuple     							=					floats_tensor((1, 3, 16, 16)     ,	rng=random.Random(UpperCamelCase_      )      ).to(UpperCamelCase_      )

          UpperCAmelCase__	:      int     							=					{
              'prompt': 'A painting of a squirrel eating a burger',
              'image': image,
              'original_image': original_image,
              'generator': generator,
              'num_inference_steps': 2,
              'output_type': 'numpy',
          }

          return inputs



      @unittest.skipIf(
          torch_device != 'cuda' or not is_xformers_available()     ,	reason='XFormers attention is only available with CUDA and `xformers` installed'     ,	)
      def  __snake_case							(   self      ):
          self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3      )



      def  __snake_case							(   self      ):
          self._test_save_load_optional_components()



      @unittest.skipIf(torch_device != 'cuda'     ,	reason='float16 requires CUDA'      )
      def  __snake_case							(   self      ):
          # Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
          super().test_save_load_floataa(expected_max_diff=1E-1      )



      def  __snake_case							(   self      ):
          self._test_attention_slicing_forward_pass(expected_max_diff=1E-2      )



      def  __snake_case							(   self      ):
          self._test_save_load_local()



      def  __snake_case							(   self      ):
          self._test_inference_batch_single_identical(
              expected_max_diff=1E-2     ,	)

