'''simple docstring'''







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

UpperCamelCase_      	=	logging.get_logger(__name__)




class _SCREAMING_SNAKE_CASE(							enum.Enum					):
						A_			:							int     		= 0
						A_			:							int     		= 1




@add_end_docstrings(_SCREAMING_SNAKE_CASE					)
class _SCREAMING_SNAKE_CASE(							_SCREAMING_SNAKE_CASE					):
						A_			:							str     		= 'generated'




						def __init__(    self :       Tuple							,							*UpperCamelCase_ :       str							,							**UpperCamelCase_ :       List[str]      )     ->				List[Any]:
													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      __lowerCamelCase       (    self :       Dict							,							UpperCamelCase_ :       Any=None							,							UpperCamelCase_ :       Union[str, Any]=None							,							UpperCamelCase_ :       str=None							,							UpperCamelCase_ :       str=None							,							UpperCamelCase_ :       Optional[int]=None							,							UpperCamelCase_ :       Optional[int]=None							,							**UpperCamelCase_ :       Union[str, Any]							,							)     ->				Union[str, Any]:
													SCREAMING_SNAKE_CASE__     :Union[str, Any]						    = {}
													if truncation is not None:
																				SCREAMING_SNAKE_CASE__     :Any						    = truncation

													SCREAMING_SNAKE_CASE__     :Optional[Any]						    = generate_kwargs

													SCREAMING_SNAKE_CASE__     :Optional[int]						    = {}
													if return_tensors is not None and return_type is None:
																				SCREAMING_SNAKE_CASE__     :List[str]						    = ReturnType.TENSORS if return_tensors else ReturnType.TEXT
													if return_type is not None:
																				SCREAMING_SNAKE_CASE__     :Tuple						    = return_type

													if clean_up_tokenization_spaces is not None:
																				SCREAMING_SNAKE_CASE__     :Optional[int]						    = clean_up_tokenization_spaces

													if stop_sequence is not None:
																				SCREAMING_SNAKE_CASE__     :Tuple						    = 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.'      )
																				SCREAMING_SNAKE_CASE__     :Optional[int]						    = stop_sequence_ids[0]

													return preprocess_params, forward_params, postprocess_params




						def      __lowerCamelCase       (    self :       List[str]							,							UpperCamelCase_ :       int							,							UpperCamelCase_ :       int							,							UpperCamelCase_ :       int      )     ->				Tuple:
													return True




						def      __lowerCamelCase       (    self :       Any							,							*UpperCamelCase_ :       Dict							,							UpperCamelCase_ :       Tuple      )     ->				List[str]:
													SCREAMING_SNAKE_CASE__     :Union[str, Any]						    = 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'      )
																				SCREAMING_SNAKE_CASE__     :str						    = ([prefix + arg for arg in args[0]],)
																				SCREAMING_SNAKE_CASE__     :List[Any]						    = True

													elif isinstance(args[0]							,							UpperCamelCase_      ):
																				SCREAMING_SNAKE_CASE__     :Optional[Any]						    = (prefix + args[0],)
																				SCREAMING_SNAKE_CASE__     :Optional[int]						    = False
													else:
																				raise ValueError(
																				    f''' `args[0]`: {args[0]} have the wrong format. The should be either of type `str` or type `list`'''      )
													SCREAMING_SNAKE_CASE__     :List[Any]						    = 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 :       List[str]							,							*UpperCamelCase_ :       Union[str, Any]							,							**UpperCamelCase_ :       Optional[Any]      )     ->				Union[str, Any]:
													SCREAMING_SNAKE_CASE__     :Any						    = 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      __lowerCamelCase       (    self :       Tuple							,							UpperCamelCase_ :       str							,							UpperCamelCase_ :       Any=TruncationStrategy.DO_NOT_TRUNCATE							,							**UpperCamelCase_ :       Optional[int]      )     ->				Union[str, Any]:
													SCREAMING_SNAKE_CASE__     :int						    = self._parse_and_tokenize(UpperCamelCase_							,							truncation=UpperCamelCase_							,							**UpperCamelCase_      )
													return inputs




						def      __lowerCamelCase       (    self :       Optional[int]							,							UpperCamelCase_ :       List[Any]							,							**UpperCamelCase_ :       List[str]      )     ->				List[str]:
													if self.framework == "pt":
																				SCREAMING_SNAKE_CASE__    ,      SCREAMING_SNAKE_CASE__     :Optional[Any]						    = model_inputs['input_ids'].shape
													elif self.framework == "tf":
																				SCREAMING_SNAKE_CASE__    ,      SCREAMING_SNAKE_CASE__     :Optional[int]						    = tf.shape(model_inputs['input_ids']      ).numpy()

													SCREAMING_SNAKE_CASE__     :Optional[int]						    = generate_kwargs.get('min_length'							,							self.model.config.min_length      )
													SCREAMING_SNAKE_CASE__     :Optional[Any]						    = generate_kwargs.get('max_length'							,							self.model.config.max_length      )
													self.check_inputs(UpperCamelCase_							,							generate_kwargs['min_length']							,							generate_kwargs['max_length']      )
													SCREAMING_SNAKE_CASE__     :Optional[Any]						    = self.model.generate(**UpperCamelCase_							,							**UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :Tuple						    = output_ids.shape[0]
													if self.framework == "pt":
																				SCREAMING_SNAKE_CASE__     :Dict						    = output_ids.reshape(UpperCamelCase_							,							out_b // in_b							,							*output_ids.shape[1:]      )
													elif self.framework == "tf":
																				SCREAMING_SNAKE_CASE__     :List[str]						    = tf.reshape(UpperCamelCase_							,							(in_b, out_b // in_b, *output_ids.shape[1:])      )
													return {"output_ids": output_ids}




						def      __lowerCamelCase       (    self :       Optional[int]							,							UpperCamelCase_ :       Any							,							UpperCamelCase_ :       List[str]=ReturnType.TEXT							,							UpperCamelCase_ :       List[str]=False      )     ->				List[str]:
													SCREAMING_SNAKE_CASE__     :Union[str, Any]						    = []
													for output_ids in model_outputs["output_ids"][0]:
																				if return_type == ReturnType.TENSORS:
																											SCREAMING_SNAKE_CASE__     :List[str]						    = {f'''{self.return_name}_token_ids''': output_ids}
																				elif return_type == ReturnType.TEXT:
																											SCREAMING_SNAKE_CASE__     :Optional[Any]						    = {
																											    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(_SCREAMING_SNAKE_CASE					)
class _SCREAMING_SNAKE_CASE(							_SCREAMING_SNAKE_CASE					):
						A_			:							List[str]     		= 'summary'




						def __call__(    self :       Union[str, Any]							,							*UpperCamelCase_ :       Any							,							**UpperCamelCase_ :       List[Any]      )     ->				Dict:
													return super().__call__(*UpperCamelCase_							,							**UpperCamelCase_      )




						def      __lowerCamelCase       (    self :       Union[str, Any]							,							UpperCamelCase_ :       int							,							UpperCamelCase_ :       int							,							UpperCamelCase_ :       int      )     ->				bool:
													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 _SCREAMING_SNAKE_CASE(							_SCREAMING_SNAKE_CASE					):
						A_			:							List[str]     		= 'translation'




						def      __lowerCamelCase       (    self :       Dict							,							UpperCamelCase_ :       int							,							UpperCamelCase_ :       int							,							UpperCamelCase_ :       int      )     ->				str:
													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      __lowerCamelCase       (    self :       str							,							*UpperCamelCase_ :       Any							,							UpperCamelCase_ :       str=TruncationStrategy.DO_NOT_TRUNCATE							,							UpperCamelCase_ :       Optional[int]=None							,							UpperCamelCase_ :       str=None      )     ->				List[str]:
													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      __lowerCamelCase       (    self :       Union[str, Any]							,							UpperCamelCase_ :       Optional[int]=None							,							UpperCamelCase_ :       Any=None							,							**UpperCamelCase_ :       Dict      )     ->				Any:
													SCREAMING_SNAKE_CASE__    ,      SCREAMING_SNAKE_CASE__    ,      SCREAMING_SNAKE_CASE__     :Optional[int]						    = super()._sanitize_parameters(**UpperCamelCase_      )
													if src_lang is not None:
																				SCREAMING_SNAKE_CASE__     :Any						    = src_lang
													if tgt_lang is not None:
																				SCREAMING_SNAKE_CASE__     :List[Any]						    = tgt_lang
													if src_lang is None and tgt_lang is None:
																				# Backward compatibility, direct arguments use is preferred.
																				SCREAMING_SNAKE_CASE__     :Optional[Any]						    = kwargs.get('task'							,							self.task      )
																				SCREAMING_SNAKE_CASE__     :int						    = task.split('_'      )
																				if task and len(UpperCamelCase_      ) == 4:
																											# translation, XX, to YY
																											SCREAMING_SNAKE_CASE__     :Dict						    = items[1]
																											SCREAMING_SNAKE_CASE__     :Union[str, Any]						    = items[3]
													return preprocess_params, forward_params, postprocess_params




						def __call__(    self :       Optional[int]							,							*UpperCamelCase_ :       Optional[Any]							,							**UpperCamelCase_ :       Dict      )     ->				Any:
													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, LDMTextToImagePipeline, UNetaDConditionModel
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    load_numpy,
    nightly,
    require_torch_gpu,
    slow,
    torch_device,
)

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


enable_full_determinism()




class _SCREAMING_SNAKE_CASE(							_SCREAMING_SNAKE_CASE						,						unittest.TestCase					):
						A_			:							Any     		= LDMTextToImagePipeline
						A_			:							Dict     		= TEXT_TO_IMAGE_PARAMS - {
						    'negative_prompt',
						    'negative_prompt_embeds',
						    'cross_attention_kwargs',
						    'prompt_embeds',
						}
						A_			:							Optional[Any]     		= PipelineTesterMixin.required_optional_params - {
						    'num_images_per_prompt',
						    'callback',
						    'callback_steps',
						}
						A_			:							Any     		= TEXT_TO_IMAGE_BATCH_PARAMS
						A_			:							Optional[int]     		= False




						def      __lowerCamelCase       (    self :       str      )     ->				List[str]:
													torch.manual_seed(0      )
													SCREAMING_SNAKE_CASE__     :List[str]						    = 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							,							)
													SCREAMING_SNAKE_CASE__     :Optional[Any]						    = DDIMScheduler(
													    beta_start=0.0_0085							,							beta_end=0.012							,							beta_schedule='scaled_linear'							,							clip_sample=UpperCamelCase_							,							set_alpha_to_one=UpperCamelCase_							,							)
													torch.manual_seed(0      )
													SCREAMING_SNAKE_CASE__     :Tuple						    = 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      )
													SCREAMING_SNAKE_CASE__     :Union[str, Any]						    = 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=10_00							,							)
													SCREAMING_SNAKE_CASE__     :List[str]						    = CLIPTextModel(UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :List[str]						    = CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      )

													SCREAMING_SNAKE_CASE__     :List[Any]						    = {
													    'unet': unet,
													    'scheduler': scheduler,
													    'vqvae': vae,
													    'bert': text_encoder,
													    'tokenizer': tokenizer,
													}
													return components




						def      __lowerCamelCase       (    self :       Dict							,							UpperCamelCase_ :       str							,							UpperCamelCase_ :       str=0      )     ->				List[Any]:
													if str(UpperCamelCase_      ).startswith('mps'      ):
																				SCREAMING_SNAKE_CASE__     :int						    = torch.manual_seed(UpperCamelCase_      )
													else:
																				SCREAMING_SNAKE_CASE__     :Dict						    = torch.Generator(device=UpperCamelCase_      ).manual_seed(UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :Any						    = {
													    'prompt': 'A painting of a squirrel eating a burger',
													    'generator': generator,
													    'num_inference_steps': 2,
													    'guidance_scale': 6.0,
													    'output_type': 'numpy',
													}
													return inputs




						def      __lowerCamelCase       (    self :       List[Any]      )     ->				Optional[int]:
													SCREAMING_SNAKE_CASE__     :Optional[int]						    = 'cpu'  # ensure determinism for the device-dependent torch.Generator

													SCREAMING_SNAKE_CASE__     :Union[str, Any]						    = self.get_dummy_components()
													SCREAMING_SNAKE_CASE__     :Tuple						    = LDMTextToImagePipeline(**UpperCamelCase_      )
													pipe.to(UpperCamelCase_      )
													pipe.set_progress_bar_config(disable=UpperCamelCase_      )

													SCREAMING_SNAKE_CASE__     :Any						    = self.get_dummy_inputs(UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :Dict						    = pipe(**UpperCamelCase_      ).images
													SCREAMING_SNAKE_CASE__     :int						    = image[0, -3:, -3:, -1]

													assert image.shape == (1, 16, 16, 3)
													SCREAMING_SNAKE_CASE__     :Optional[Any]						    = np.array([0.6101, 0.6156, 0.5622, 0.4895, 0.6661, 0.3804, 0.5748, 0.6136, 0.5014]      )

													assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-3




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




						def      __lowerCamelCase       (    self :       str      )     ->				Union[str, Any]:
													super().tearDown()
													gc.collect()
													torch.cuda.empty_cache()




						def      __lowerCamelCase       (    self :       Optional[Any]							,							UpperCamelCase_ :       Union[str, Any]							,							UpperCamelCase_ :       Optional[Any]=torch.floataa							,							UpperCamelCase_ :       int=0      )     ->				str:
													SCREAMING_SNAKE_CASE__     :str						    = torch.manual_seed(UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :Optional[int]						    = np.random.RandomState(UpperCamelCase_      ).standard_normal((1, 4, 32, 32)      )
													SCREAMING_SNAKE_CASE__     :Optional[Any]						    = torch.from_numpy(UpperCamelCase_      ).to(device=UpperCamelCase_							,							dtype=UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :Union[str, Any]						    = {
													    'prompt': 'A painting of a squirrel eating a burger',
													    'latents': latents,
													    'generator': generator,
													    'num_inference_steps': 3,
													    'guidance_scale': 6.0,
													    'output_type': 'numpy',
													}
													return inputs




						def      __lowerCamelCase       (    self :       Tuple      )     ->				Any:
													SCREAMING_SNAKE_CASE__     :Optional[int]						    = LDMTextToImagePipeline.from_pretrained('CompVis/ldm-text2im-large-256'      ).to(UpperCamelCase_      )
													pipe.set_progress_bar_config(disable=UpperCamelCase_      )

													SCREAMING_SNAKE_CASE__     :Optional[int]						    = self.get_inputs(UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :Any						    = pipe(**UpperCamelCase_      ).images
													SCREAMING_SNAKE_CASE__     :Tuple						    = image[0, -3:, -3:, -1].flatten()

													assert image.shape == (1, 2_56, 2_56, 3)
													SCREAMING_SNAKE_CASE__     :str						    = np.array([0.5_1825, 0.5_2850, 0.5_2543, 0.5_4258, 0.5_2304, 0.5_2569, 0.5_4363, 0.5_5276, 0.5_6878]      )
													SCREAMING_SNAKE_CASE__     :Union[str, Any]						    = np.abs(expected_slice - image_slice      ).max()
													assert max_diff < 1e-3




@nightly
@require_torch_gpu
class _SCREAMING_SNAKE_CASE(							unittest.TestCase					):




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




						def      __lowerCamelCase       (    self :       Optional[int]							,							UpperCamelCase_ :       Any							,							UpperCamelCase_ :       Any=torch.floataa							,							UpperCamelCase_ :       List[str]=0      )     ->				str:
													SCREAMING_SNAKE_CASE__     :Union[str, Any]						    = torch.manual_seed(UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :Optional[Any]						    = np.random.RandomState(UpperCamelCase_      ).standard_normal((1, 4, 32, 32)      )
													SCREAMING_SNAKE_CASE__     :Optional[Any]						    = torch.from_numpy(UpperCamelCase_      ).to(device=UpperCamelCase_							,							dtype=UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :Optional[Any]						    = {
													    'prompt': 'A painting of a squirrel eating a burger',
													    'latents': latents,
													    'generator': generator,
													    'num_inference_steps': 50,
													    'guidance_scale': 6.0,
													    'output_type': 'numpy',
													}
													return inputs




						def      __lowerCamelCase       (    self :       int      )     ->				Dict:
													SCREAMING_SNAKE_CASE__     :str						    = LDMTextToImagePipeline.from_pretrained('CompVis/ldm-text2im-large-256'      ).to(UpperCamelCase_      )
													pipe.set_progress_bar_config(disable=UpperCamelCase_      )

													SCREAMING_SNAKE_CASE__     :List[Any]						    = self.get_inputs(UpperCamelCase_      )
													SCREAMING_SNAKE_CASE__     :Any						    = pipe(**UpperCamelCase_      ).images[0]

													SCREAMING_SNAKE_CASE__     :List[Any]						    = load_numpy(
													    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/ldm_text2img/ldm_large_256_ddim.npy'      )
													SCREAMING_SNAKE_CASE__     :str						    = np.abs(expected_image - image      ).max()
													assert max_diff < 1e-3






