


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

_lowercase      :      List[Any]            =					logging.get_logger(__name__)






class 	UpperCamelCase__(       enum.Enum	):
				__magic_name__    :				Union[str, Any]  		=	0
				__magic_name__    :				Dict  		=	1







@add_end_docstrings(_UpperCamelCase	)
class 	UpperCamelCase__(       _UpperCamelCase	):
				__magic_name__    :				int  		=	'generated'




				def __init__(		self      :			List[Any]				,   *lowerCAmelCase      :			Any				,   **lowerCAmelCase      :			Optional[int]				)->					Any:

										"""simple docstring"""




										super().__init__(*_UpperCAmelCase				,   **_UpperCAmelCase				)

										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 					a__(		self      :			Optional[int]				,   lowerCAmelCase      :			List[Any]=None				,   lowerCAmelCase      :			Tuple=None				,   lowerCAmelCase      :			List[Any]=None				,   lowerCAmelCase      :			Union[str, Any]=None				,   lowerCAmelCase      :			Dict=None				,   lowerCAmelCase      :			Optional[int]=None				,   **lowerCAmelCase      :			List[Any]				,   )->					List[str]:

										"""simple docstring"""




										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(_UpperCAmelCase				,   add_special_tokens=_UpperCAmelCase				)
																if len(_UpperCAmelCase				) > 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 					a__(		self      :			Any				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			int				)->					str:

										"""simple docstring"""




										return True




				def 					a__(		self      :			str				,   *lowerCAmelCase      :			Any				,   lowerCAmelCase      :			Union[str, Any]				)->					Any:

										"""simple docstring"""




										UpperCAmelCase					    =     self.model.config.prefix if self.model.config.prefix is not None else ''
										if isinstance(args[0]				,   _UpperCAmelCase				):
																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]				,   _UpperCAmelCase				):
																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(*_UpperCAmelCase				,   padding=_UpperCAmelCase				,   truncation=_UpperCAmelCase				,   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[int]				,   *lowerCAmelCase      :			List[Any]				,   **lowerCAmelCase      :			Any				)->					List[Any]:

										"""simple docstring"""




										UpperCAmelCase					    =     super().__call__(*_UpperCAmelCase				,   **_UpperCAmelCase				)
										if (
										    isinstance(args[0]				,   _UpperCAmelCase				)
										    and all(isinstance(_UpperCAmelCase				,   _UpperCAmelCase				) for el in args[0]				)
										    and all(len(_UpperCAmelCase				) == 1 for res in result				)
										):
																return [res[0] for res in result]
										return result




				def 					a__(		self      :			Tuple				,   lowerCAmelCase      :			Union[str, Any]				,   lowerCAmelCase      :			List[str]=TruncationStrategy.DO_NOT_TRUNCATE				,   **lowerCAmelCase      :			int				)->					Optional[Any]:

										"""simple docstring"""




										UpperCAmelCase					    =     self._parse_and_tokenize(_UpperCAmelCase				,   truncation=_UpperCAmelCase				,   **_UpperCAmelCase				)
										return inputs




				def 					a__(		self      :			Any				,   lowerCAmelCase      :			str				,   **lowerCAmelCase      :			Tuple				)->					Tuple:

										"""simple docstring"""




										if self.framework == "pt":
																UpperCAmelCase					    =     model_inputs['input_ids'].shape
										elif self.framework == "tf":
																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(_UpperCAmelCase				,   generate_kwargs['''min_length''']				,   generate_kwargs['''max_length''']				)
										UpperCAmelCase					    =     self.model.generate(**_UpperCAmelCase				,   **_UpperCAmelCase				)
										UpperCAmelCase					    =     output_ids.shape[0]
										if self.framework == "pt":
																UpperCAmelCase					    =     output_ids.reshape(_UpperCAmelCase				,   out_b // in_b				,   *output_ids.shape[1:]				)
										elif self.framework == "tf":
																UpperCAmelCase					    =     tf.reshape(_UpperCAmelCase				,   (in_b, out_b // in_b, *output_ids.shape[1:])				)
										return {"output_ids": output_ids}





				def 					a__(		self      :			Dict				,   lowerCAmelCase      :			Tuple				,   lowerCAmelCase      :			Any=ReturnType.TEXT				,   lowerCAmelCase      :			Dict=False				)->					Dict:

										"""simple docstring"""




										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(
																						        _UpperCAmelCase				,   skip_special_tokens=_UpperCAmelCase				,   clean_up_tokenization_spaces=_UpperCAmelCase				,   )
																						}
																records.append(_UpperCAmelCase				)
										return records







@add_end_docstrings(_UpperCamelCase	)
class 	UpperCamelCase__(       _UpperCamelCase	):
				__magic_name__    :				Any  		=	'summary'




				def __call__(		self      :			Any				,   *lowerCAmelCase      :			Union[str, Any]				,   **lowerCAmelCase      :			List[str]				)->					Optional[int]:

										"""simple docstring"""




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





				def 					a__(		self      :			Optional[Any]				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			int				)->					Optional[Any]:

										"""simple docstring"""




										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(_UpperCamelCase	)
class 	UpperCamelCase__(       _UpperCamelCase	):
				__magic_name__    :				List[Any]  		=	'translation'




				def 					a__(		self      :			Optional[int]				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			int				)->					str:

										"""simple docstring"""




										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 					a__(		self      :			Any				,   *lowerCAmelCase      :			Any				,   lowerCAmelCase      :			List[str]=TruncationStrategy.DO_NOT_TRUNCATE				,   lowerCAmelCase      :			Optional[int]=None				,   lowerCAmelCase      :			Any=None				)->					Any:

										"""simple docstring"""




										if getattr(self.tokenizer				,   '''_build_translation_inputs'''				,   _UpperCAmelCase				):
																return self.tokenizer._build_translation_inputs(
																    *_UpperCAmelCase				,   return_tensors=self.framework				,   truncation=_UpperCAmelCase				,   src_lang=_UpperCAmelCase				,   tgt_lang=_UpperCAmelCase				)
										else:
																return super()._parse_and_tokenize(*_UpperCAmelCase				,   truncation=_UpperCAmelCase				)




				def 					a__(		self      :			Tuple				,   lowerCAmelCase      :			List[Any]=None				,   lowerCAmelCase      :			List[str]=None				,   **lowerCAmelCase      :			Dict				)->					List[str]:

										"""simple docstring"""




										UpperCAmelCase					    =     super()._sanitize_parameters(**_UpperCAmelCase				)
										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(_UpperCAmelCase				) == 4:
																						# translation, XX, to YY
																						UpperCAmelCase					    =     items[1]
																						UpperCAmelCase					    =     items[3]
										return preprocess_params, forward_params, postprocess_params





				def __call__(		self      :			Optional[Any]				,   *lowerCAmelCase      :			Tuple				,   **lowerCAmelCase      :			Optional[int]				)->					int:

										"""simple docstring"""




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









'''simple docstring'''



import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class 	UpperCamelCase__(       lowerCAmelCase						,    lowerCAmelCase						,    unittest.TestCase	):
				__magic_name__    :				List[str]  		=	StableDiffusionSAGPipeline
				__magic_name__    :				str  		=	TEXT_TO_IMAGE_PARAMS
				__magic_name__    :				Any  		=	TEXT_TO_IMAGE_BATCH_PARAMS
				__magic_name__    :				Optional[int]  		=	TEXT_TO_IMAGE_IMAGE_PARAMS
				__magic_name__    :				List[Any]  		=	TEXT_TO_IMAGE_IMAGE_PARAMS
				__magic_name__    :				str  		=	False




				def 					a__(		self      :			Union[str, Any]				)->					Tuple:

										"""simple docstring"""




										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''')				,   cross_attention_dim=32				,   )
										UpperCAmelCase					    =     DDIMScheduler(
										    beta_start=0.00085				,   beta_end=0.012				,   beta_schedule='''scaled_linear'''				,   clip_sample=lowerCAmelCase				,   set_alpha_to_one=lowerCAmelCase				,   )
										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				,   )
										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				,   )
										UpperCAmelCase					    =     CLIPTextModel(lowerCAmelCase				)
										UpperCAmelCase					    =     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''				)

										UpperCAmelCase					    =     {
										    '''unet''': unet,
										    '''scheduler''': scheduler,
										    '''vae''': vae,
										    '''text_encoder''': text_encoder,
										    '''tokenizer''': tokenizer,
										    '''safety_checker''': None,
										    '''feature_extractor''': None,
										}
										return components




				def 					a__(		self      :			Optional[Any]				,   lowerCAmelCase      :			str				,   lowerCAmelCase      :			Tuple=0				)->					str:

										"""simple docstring"""




										if str(lowerCAmelCase				).startswith('''mps'''				):
																UpperCAmelCase					    =     torch.manual_seed(lowerCAmelCase				)
										else:
																UpperCAmelCase					    =     torch.Generator(device=lowerCAmelCase				).manual_seed(lowerCAmelCase				)
										UpperCAmelCase					    =     {
										    '''prompt''': '''.''',
										    '''generator''': generator,
										    '''num_inference_steps''': 2,
										    '''guidance_scale''': 1.0,
										    '''sag_scale''': 1.0,
										    '''output_type''': '''numpy''',
										}
										return inputs





				def 					a__(		self      :			Any				)->					List[str]:

										"""simple docstring"""




										super().test_inference_batch_single_identical(expected_max_diff=3E-3				)







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




				def 					a__(		self      :			Union[str, Any]				)->					Any:

										"""simple docstring"""




										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()




				def 					a__(		self      :			Union[str, Any]				)->					Tuple:

										"""simple docstring"""




										UpperCAmelCase					    =     StableDiffusionSAGPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''				)
										UpperCAmelCase					    =     sag_pipe.to(lowerCAmelCase				)
										sag_pipe.set_progress_bar_config(disable=lowerCAmelCase				)

										UpperCAmelCase					    =     '''.'''
										UpperCAmelCase					    =     torch.manual_seed(0				)
										UpperCAmelCase					    =     sag_pipe(
										    [prompt]				,   generator=lowerCAmelCase				,   guidance_scale=7.5				,   sag_scale=1.0				,   num_inference_steps=20				,   output_type='''np'''				)

										UpperCAmelCase					    =     output.images

										UpperCAmelCase					    =     image[0, -3:, -3:, -1]

										assert image.shape == (1, 512, 512, 3)
										UpperCAmelCase					    =     np.array([0.1568, 0.1738, 0.1695, 0.1693, 0.1507, 0.1705, 0.1547, 0.1751, 0.1949]				)

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




				def 					a__(		self      :			int				)->					Union[str, Any]:

										"""simple docstring"""




										UpperCAmelCase					    =     StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''				)
										UpperCAmelCase					    =     sag_pipe.to(lowerCAmelCase				)
										sag_pipe.set_progress_bar_config(disable=lowerCAmelCase				)

										UpperCAmelCase					    =     '''.'''
										UpperCAmelCase					    =     torch.manual_seed(0				)
										UpperCAmelCase					    =     sag_pipe(
										    [prompt]				,   generator=lowerCAmelCase				,   guidance_scale=7.5				,   sag_scale=1.0				,   num_inference_steps=20				,   output_type='''np'''				)

										UpperCAmelCase					    =     output.images

										UpperCAmelCase					    =     image[0, -3:, -3:, -1]

										assert image.shape == (1, 512, 512, 3)
										UpperCAmelCase					    =     np.array([0.3459, 0.2876, 0.2537, 0.3002, 0.2671, 0.2160, 0.3026, 0.2262, 0.2371]				)

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





				def 					a__(		self      :			Optional[Any]				)->					Union[str, Any]:

										"""simple docstring"""




										UpperCAmelCase					    =     StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''				)
										UpperCAmelCase					    =     sag_pipe.to(lowerCAmelCase				)
										sag_pipe.set_progress_bar_config(disable=lowerCAmelCase				)

										UpperCAmelCase					    =     '''.'''
										UpperCAmelCase					    =     torch.manual_seed(0				)
										UpperCAmelCase					    =     sag_pipe(
										    [prompt]				,   width=768				,   height=512				,   generator=lowerCAmelCase				,   guidance_scale=7.5				,   sag_scale=1.0				,   num_inference_steps=20				,   output_type='''np'''				,   )

										UpperCAmelCase					    =     output.images

										assert image.shape == (1, 512, 768, 3)





