


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












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   lowerCamelCase      (_snake_case ,  _snake_case ,  unittest.TestCase			):







		'''simple docstring'''



		_snake_case    :    Union[str, Any]       						=  IFImgaImgSuperResolutionPipeline
		_snake_case    :    Union[str, Any]       						=  TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''width''', '''height'''}
		_snake_case    :    List[Any]       						=  TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''original_image'''}			)
		_snake_case    :    List[str]       						=  PipelineTesterMixin.required_optional_params - {'''latents'''}



		def 					__UpperCAmelCase			(							self   )	->							Optional[Any]:
							return self._get_superresolution_dummy_components()



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		_UpperCamelCase=0   )	->							Any:
							if str(_UpperCamelCase   ).startswith('mps'   ):
												UpperCAmelCase_    :      List[Any]            =						torch.manual_seed(_UpperCamelCase   )
							else:
												UpperCAmelCase_    :      int            =						torch.Generator(device=_UpperCamelCase   ).manual_seed(_UpperCamelCase   )

							UpperCAmelCase_    :      List[Any]            =						floats_tensor((1, 3, 3_2, 3_2)				,		rng=random.Random(_UpperCamelCase   )   ).to(_UpperCamelCase   )
							UpperCAmelCase_    :      Dict            =						floats_tensor((1, 3, 1_6, 1_6)				,		rng=random.Random(_UpperCamelCase   )   ).to(_UpperCamelCase   )

							UpperCAmelCase_    :      Tuple            =						{
							    '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 					__UpperCAmelCase			(							self   )	->							Any:
							self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3   )



		def 					__UpperCAmelCase			(							self   )	->							Dict:
							self._test_save_load_optional_components()



		@unittest.skipIf(torch_device != 'cuda'				,		reason='float16 requires CUDA'   )
		def 					__UpperCAmelCase			(							self   )	->							str:
							# 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 					__UpperCAmelCase			(							self   )	->							List[Any]:
							self._test_attention_slicing_forward_pass(expected_max_diff=1E-2   )



		def 					__UpperCAmelCase			(							self   )	->							Union[str, Any]:
							self._test_save_load_local()



		def 					__UpperCAmelCase			(							self   )	->							Dict:
							self._test_inference_batch_single_identical(
							    expected_max_diff=1E-2				,		)






