




import enum
import warnings

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
 import tensorflow as tf

 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

a_   		=     logging.get_logger(__name__)

class  _lowerCamelCase						(			enum.Enum   ):

      """simple docstring"""

      lowerCAmelCase__			: Union[str, Any]    =					0
      lowerCAmelCase__			: Optional[Any]    =					1







@add_end_docstrings(_SCREAMING_SNAKE_CASE   )
class  _lowerCamelCase						(			_SCREAMING_SNAKE_CASE   ):

      """simple docstring"""

      lowerCAmelCase__			: Optional[int]    =					"generated"







      def __init__(     self					: Optional[int] ,       *snake_case					: Optional[Any] ,       **snake_case					: List[Any]				):
          super().__init__(*snake_case ,       **snake_case				)

          self.check_model_type(
              TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
              if self.framework == '''tf'''
              else MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING				)







      def       snake_case	(     self					: Any ,       snake_case					: Optional[Any]=None ,       snake_case					: Any=None ,       snake_case					: List[str]=None ,       snake_case					: int=None ,       snake_case					: Union[str, Any]=None ,       snake_case					: str=None ,       **snake_case					: Optional[Any] ,       ):
          __UpperCamelCase  		=						{}
          if truncation is not None:
              __UpperCamelCase  		=						truncation

          __UpperCamelCase  		=						generate_kwargs

          __UpperCamelCase  		=						{}
          if return_tensors is not None and return_type is None:
              __UpperCamelCase  		=						ReturnType.TENSORS if return_tensors else ReturnType.TEXT
          if return_type is not None:
              __UpperCamelCase  		=						return_type

          if clean_up_tokenization_spaces is not None:
              __UpperCamelCase  		=						clean_up_tokenization_spaces

          if stop_sequence is not None:
              __UpperCamelCase  		=						self.tokenizer.encode(snake_case ,       add_special_tokens=snake_case				)
              if len(snake_case				) > 1:
                  warnings.warn(
                      '''Stopping on a multiple token sequence is not yet supported on transformers. The first token of'''
                      ''' the stop sequence will be used as the stop sequence string in the interim.'''				)
              __UpperCamelCase  		=						stop_sequence_ids[0]

          return preprocess_params, forward_params, postprocess_params







      def       snake_case	(     self					: List[str] ,       snake_case					: int ,       snake_case					: int ,       snake_case					: int				):
          return True







      def       snake_case	(     self					: List[Any] ,       *snake_case					: Tuple ,       snake_case					: Optional[int]				):
          __UpperCamelCase  		=						self.model.config.prefix if self.model.config.prefix is not None else ''''''
          if isinstance(args[0] ,       snake_case				):
              if self.tokenizer.pad_token_id is None:
                  raise ValueError('''Please make sure that the tokenizer has a pad_token_id when using a batch input'''				)
              __UpperCamelCase  		=						([prefix + arg for arg in args[0]],)
              __UpperCamelCase  		=						True

          elif isinstance(args[0] ,       snake_case				):
              __UpperCamelCase  		=						(prefix + args[0],)
              __UpperCamelCase  		=						False
          else:
              raise ValueError(
                  F" `args[0]`: {args[0]} have the wrong format. The should be either of type `str` or type `list`"				)
          __UpperCamelCase  		=						self.tokenizer(*snake_case ,       padding=snake_case ,       truncation=snake_case ,       return_tensors=self.framework				)
          # This is produced by tokenizers but is an invalid generate kwargs
          if "token_type_ids" in inputs:
              del inputs["token_type_ids"]
          return inputs







      def __call__(     self					: Optional[Any] ,       *snake_case					: Optional[int] ,       **snake_case					: Optional[int]				):
          __UpperCamelCase  		=						super().__call__(*snake_case ,       **snake_case				)
          if (
              isinstance(args[0] ,       snake_case				)
              and all(isinstance(snake_case ,       snake_case				) for el in args[0]				)
              and all(len(snake_case				) == 1 for res in result				)
          ):
              return [res[0] for res in result]
          return result







      def       snake_case	(     self					: List[Any] ,       snake_case					: str ,       snake_case					: List[str]=TruncationStrategy.DO_NOT_TRUNCATE ,       **snake_case					: Optional[int]				):
          __UpperCamelCase  		=						self._parse_and_tokenize(snake_case ,       truncation=snake_case ,       **snake_case				)
          return inputs







      def       snake_case	(     self					: Optional[int] ,       snake_case					: Any ,       **snake_case					: List[Any]				):
          if self.framework == "pt":
              __UpperCamelCase		,      __UpperCamelCase  		=						model_inputs['''input_ids'''].shape
          elif self.framework == "tf":
              __UpperCamelCase		,      __UpperCamelCase  		=						tf.shape(model_inputs['''input_ids''']				).numpy()

          __UpperCamelCase  		=						generate_kwargs.get('''min_length''' ,       self.model.config.min_length				)
          __UpperCamelCase  		=						generate_kwargs.get('''max_length''' ,       self.model.config.max_length				)
          self.check_inputs(snake_case ,       generate_kwargs['''min_length'''] ,       generate_kwargs['''max_length''']				)
          __UpperCamelCase  		=						self.model.generate(**snake_case ,       **snake_case				)
          __UpperCamelCase  		=						output_ids.shape[0]
          if self.framework == "pt":
              __UpperCamelCase  		=						output_ids.reshape(snake_case ,       out_b // in_b ,       *output_ids.shape[1:]				)
          elif self.framework == "tf":
              __UpperCamelCase  		=						tf.reshape(snake_case ,       (in_b, out_b // in_b, *output_ids.shape[1:])				)
          return {"output_ids": output_ids}







      def       snake_case	(     self					: Optional[int] ,       snake_case					: str ,       snake_case					: Dict=ReturnType.TEXT ,       snake_case					: str=False				):
          __UpperCamelCase  		=						[]
          for output_ids in model_outputs["output_ids"][0]:
              if return_type == ReturnType.TENSORS:
                  __UpperCamelCase  		=						{F"{self.return_name}_token_ids": output_ids}
              elif return_type == ReturnType.TEXT:
                  __UpperCamelCase  		=						{
                      F"{self.return_name}_text": self.tokenizer.decode(
                          snake_case ,       skip_special_tokens=snake_case ,       clean_up_tokenization_spaces=snake_case ,       )
                  }
              records.append(snake_case				)
          return records







@add_end_docstrings(_SCREAMING_SNAKE_CASE   )
class  _lowerCamelCase						(			_SCREAMING_SNAKE_CASE   ):

      """simple docstring"""

      lowerCAmelCase__			: List[Any]    =					"summary"







      def __call__(     self					: Optional[Any] ,       *snake_case					: List[str] ,       **snake_case					: Optional[Any]				):
          return super().__call__(*snake_case ,       **snake_case				)







      def       snake_case	(     self					: Tuple ,       snake_case					: int ,       snake_case					: int ,       snake_case					: int				):
          if max_length < min_length:
              logger.warning(F"Your min_length={min_length} must be inferior than your max_length={max_length}."				)

          if input_length < max_length:
              logger.warning(
                  F"Your max_length is set to {max_length}, but your input_length is only {input_length}. Since this is "
                  '''a summarization task, where outputs shorter than the input are typically wanted, you might '''
                  F"consider decreasing max_length manually, e.g. summarizer('...', max_length={input_length//2})"				)







@add_end_docstrings(_SCREAMING_SNAKE_CASE   )
class  _lowerCamelCase						(			_SCREAMING_SNAKE_CASE   ):

      """simple docstring"""

      lowerCAmelCase__			: Union[str, Any]    =					"translation"







      def       snake_case	(     self					: Tuple ,       snake_case					: int ,       snake_case					: int ,       snake_case					: int				):
          if input_length > 0.9 * max_length:
              logger.warning(
                  F"Your input_length: {input_length} is bigger than 0.9 * max_length: {max_length}. You might consider "
                  '''increasing your max_length manually, e.g. translator(\'...\', max_length=400)'''				)
          return True







      def       snake_case	(     self					: str ,       *snake_case					: Union[str, Any] ,       snake_case					: Dict=TruncationStrategy.DO_NOT_TRUNCATE ,       snake_case					: Tuple=None ,       snake_case					: Tuple=None				):
          if getattr(self.tokenizer ,       '''_build_translation_inputs''' ,       snake_case				):
              return self.tokenizer._build_translation_inputs(
                  *snake_case ,       return_tensors=self.framework ,       truncation=snake_case ,       src_lang=snake_case ,       tgt_lang=snake_case				)
          else:
              return super()._parse_and_tokenize(*snake_case ,       truncation=snake_case				)







      def       snake_case	(     self					: Any ,       snake_case					: Dict=None ,       snake_case					: Optional[int]=None ,       **snake_case					: str				):
          __UpperCamelCase		,      __UpperCamelCase		,      __UpperCamelCase  		=						super()._sanitize_parameters(**snake_case				)
          if src_lang is not None:
              __UpperCamelCase  		=						src_lang
          if tgt_lang is not None:
              __UpperCamelCase  		=						tgt_lang
          if src_lang is None and tgt_lang is None:
              # Backward compatibility, direct arguments use is preferred.
              __UpperCamelCase  		=						kwargs.get('''task''' ,       self.task				)
              __UpperCamelCase  		=						task.split('''_'''				)
              if task and len(snake_case				) == 4:
                  # translation, XX, to YY
                  __UpperCamelCase  		=						items[1]
                  __UpperCamelCase  		=						items[3]
          return preprocess_params, forward_params, postprocess_params







      def __call__(     self					: Any ,       *snake_case					: Union[str, Any] ,       **snake_case					: Optional[Any]				):
          return super().__call__(*snake_case ,       **snake_case				)












import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class  _lowerCamelCase						(			_SCREAMING_SNAKE_CASE							,			_SCREAMING_SNAKE_CASE							,			unittest.TestCase   ):

      """simple docstring"""

      lowerCAmelCase__			: Optional[int]    =					StableDiffusionXLImgaImgPipeline
      lowerCAmelCase__			: List[str]    =					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
      lowerCAmelCase__			: int    =					PipelineTesterMixin.required_optional_params - {"latents"}
      lowerCAmelCase__			: Tuple    =					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
      lowerCAmelCase__			: int    =					IMAGE_TO_IMAGE_IMAGE_PARAMS
      lowerCAmelCase__			: List[Any]    =					IMAGE_TO_IMAGE_IMAGE_PARAMS







      def       snake_case	(     self					: Union[str, Any]				):
          torch.manual_seed(0				)
          __UpperCamelCase  		=						UNetaDConditionModel(
              block_out_channels=(32, 64) ,       layers_per_block=2 ,       sample_size=32 ,       in_channels=4 ,       out_channels=4 ,       down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''') ,       up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''') ,       attention_head_dim=(2, 4) ,       use_linear_projection=snake_case ,       addition_embed_type='''text_time''' ,       addition_time_embed_dim=8 ,       transformer_layers_per_block=(1, 2) ,       projection_class_embeddings_input_dim=80 ,       cross_attention_dim=64 ,       )
          __UpperCamelCase  		=						EulerDiscreteScheduler(
              beta_start=0.0_00_85 ,       beta_end=0.0_12 ,       steps_offset=1 ,       beta_schedule='''scaled_linear''' ,       timestep_spacing='''leading''' ,       )
          torch.manual_seed(0				)
          __UpperCamelCase  		=						AutoencoderKL(
              block_out_channels=[32, 64] ,       in_channels=3 ,       out_channels=3 ,       down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D'''] ,       up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D'''] ,       latent_channels=4 ,       sample_size=128 ,       )
          torch.manual_seed(0				)
          __UpperCamelCase  		=						CLIPTextConfig(
              bos_token_id=0 ,       eos_token_id=2 ,       hidden_size=32 ,       intermediate_size=37 ,       layer_norm_eps=1E-05 ,       num_attention_heads=4 ,       num_hidden_layers=5 ,       pad_token_id=1 ,       vocab_size=1000 ,       hidden_act='''gelu''' ,       projection_dim=32 ,       )
          __UpperCamelCase  		=						CLIPTextModel(snake_case				)
          __UpperCamelCase  		=						CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''' ,       local_files_only=snake_case				)

          __UpperCamelCase  		=						CLIPTextModelWithProjection(snake_case				)
          __UpperCamelCase  		=						CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''' ,       local_files_only=snake_case				)

          __UpperCamelCase  		=						{
              '''unet''': unet,
              '''scheduler''': scheduler,
              '''vae''': vae,
              '''text_encoder''': text_encoder,
              '''tokenizer''': tokenizer,
              '''text_encoder_2''': text_encoder_a,
              '''tokenizer_2''': tokenizer_a,
              # "safety_checker": None,
              # "feature_extractor": None,
          }
          return components







      def       snake_case	(     self					: List[Any] ,       snake_case					: List[Any] ,       snake_case					: str=0				):
          __UpperCamelCase  		=						floats_tensor((1, 3, 32, 32) ,       rng=random.Random(snake_case				)				).to(snake_case				)
          __UpperCamelCase  		=						image / 2 + 0.5
          if str(snake_case				).startswith('''mps'''				):
              __UpperCamelCase  		=						torch.manual_seed(snake_case				)
          else:
              __UpperCamelCase  		=						torch.Generator(device=snake_case				).manual_seed(snake_case				)
          __UpperCamelCase  		=						{
              '''prompt''': '''A painting of a squirrel eating a burger''',
              '''image''': image,
              '''generator''': generator,
              '''num_inference_steps''': 2,
              '''guidance_scale''': 5.0,
              '''output_type''': '''numpy''',
              '''strength''': 0.75,
          }
          return inputs







      def       snake_case	(     self					: Tuple				):
          __UpperCamelCase  		=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
          __UpperCamelCase  		=						self.get_dummy_components()
          __UpperCamelCase  		=						StableDiffusionXLImgaImgPipeline(**snake_case				)
          __UpperCamelCase  		=						sd_pipe.to(snake_case				)
          sd_pipe.set_progress_bar_config(disable=snake_case				)

          __UpperCamelCase  		=						self.get_dummy_inputs(snake_case				)
          __UpperCamelCase  		=						sd_pipe(**snake_case				).images
          __UpperCamelCase  		=						image[0, -3:, -3:, -1]

          assert image.shape == (1, 32, 32, 3)

          __UpperCamelCase  		=						np.array([0.46_56, 0.48_40, 0.44_39, 0.66_98, 0.55_74, 0.45_24, 0.57_99, 0.59_43, 0.51_65]				)

          assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2







      def       snake_case	(     self					: Optional[int]				):
          super().test_attention_slicing_forward_pass(expected_max_diff=3E-3				)







      def       snake_case	(     self					: Optional[Any]				):
          super().test_inference_batch_single_identical(expected_max_diff=3E-3				)







      def       snake_case	(     self					: Optional[Any]				):
          pass







      def       snake_case	(     self					: str				):
          __UpperCamelCase  		=						self.get_dummy_components()
          __UpperCamelCase  		=						StableDiffusionXLImgaImgPipeline(**snake_case				)
          __UpperCamelCase  		=						sd_pipe.to(snake_case				)
          __UpperCamelCase  		=						sd_pipe.to(snake_case				)
          sd_pipe.set_progress_bar_config(disable=snake_case				)

          # forward without prompt embeds
          __UpperCamelCase  		=						self.get_dummy_inputs(snake_case				)
          __UpperCamelCase  		=						3 * ['''this is a negative prompt''']
          __UpperCamelCase  		=						negative_prompt
          __UpperCamelCase  		=						3 * [inputs['''prompt''']]

          __UpperCamelCase  		=						sd_pipe(**snake_case				)
          __UpperCamelCase  		=						output.images[0, -3:, -3:, -1]

          # forward with prompt embeds
          __UpperCamelCase  		=						self.get_dummy_inputs(snake_case				)
          __UpperCamelCase  		=						3 * ['''this is a negative prompt''']
          __UpperCamelCase  		=						3 * [inputs.pop('''prompt'''				)]

          (
              (
              __UpperCamelCase
          )		,      (
              __UpperCamelCase
          )		,      (
              __UpperCamelCase
          )		,      (
              __UpperCamelCase
          )		,      
          )  		=						sd_pipe.encode_prompt(snake_case ,       negative_prompt=snake_case				)

          __UpperCamelCase  		=						sd_pipe(
              **snake_case ,       prompt_embeds=snake_case ,       negative_prompt_embeds=snake_case ,       pooled_prompt_embeds=snake_case ,       negative_pooled_prompt_embeds=snake_case ,       )
          __UpperCamelCase  		=						output.images[0, -3:, -3:, -1]

          # make sure that it's equal
          assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()				).max() < 1E-4







@slow
@require_torch_gpu
class  _lowerCamelCase						(			unittest.TestCase   ):

      """simple docstring"""

      def       snake_case	(     self					: Optional[Any]				):
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()







      def       snake_case	(     self					: str ,       snake_case					: List[Any] ,       snake_case					: str="cpu" ,       snake_case					: int=torch.floataa ,       snake_case					: Optional[int]=0				):
          __UpperCamelCase  		=						torch.Generator(device=snake_case				).manual_seed(snake_case				)
          __UpperCamelCase  		=						np.random.RandomState(snake_case				).standard_normal((1, 4, 64, 64)				)
          __UpperCamelCase  		=						torch.from_numpy(snake_case				).to(device=snake_case ,       dtype=snake_case				)
          __UpperCamelCase  		=						{
              '''prompt''': '''a photograph of an astronaut riding a horse''',
              '''latents''': latents,
              '''generator''': generator,
              '''num_inference_steps''': 3,
              '''guidance_scale''': 7.5,
              '''output_type''': '''numpy''',
          }
          return inputs







      def       snake_case	(     self					: Dict				):
          __UpperCamelCase  		=						DiffusionPipeline.from_pretrained('''stabilityai/stable-diffusion-2-base'''				)
          pipe.to(snake_case				)
          pipe.set_progress_bar_config(disable=snake_case				)

          __UpperCamelCase  		=						self.get_inputs(snake_case				)
          __UpperCamelCase  		=						pipe(**snake_case				).images
          __UpperCamelCase  		=						image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 512, 512, 3)
          __UpperCamelCase  		=						np.array([0.4_94_93, 0.4_78_96, 0.4_07_98, 0.5_42_14, 0.5_32_12, 0.4_82_02, 0.4_76_56, 0.4_63_29, 0.4_85_06]				)
          assert np.abs(image_slice - expected_slice				).max() < 7E-3






