


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

__snake_case    						=logging.get_logger(__name__)

class 		UpperCAmelCase_      (  enum.Enum		):
     lowerCamelCase				:	Optional[int]		=     0
     lowerCamelCase				:	str		=     1




@add_end_docstrings(__lowercase		)
class 		UpperCAmelCase_      (  __lowercase		):
     lowerCamelCase				:	Tuple		=     '''generated'''





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

        lowerCAmelCase     =			generate_kwargs

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

        if clean_up_tokenization_spaces is not None:
           lowerCAmelCase     =			clean_up_tokenization_spaces

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

        return preprocess_params, forward_params, postprocess_params





     def 							__UpperCAmelCase      ( self       :      List[Any]    ,      UpperCAmelCase__       :      int    ,      UpperCAmelCase__       :      int    ,      UpperCAmelCase__       :      int    )				->  List[Any]:
        return True





     def 							__UpperCAmelCase      ( self       :      Optional[int]    ,      *UpperCAmelCase__       :      Optional[int]    ,      UpperCAmelCase__       :      List[Any]    )				->  Optional[int]:
        lowerCAmelCase     =			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'    )
           lowerCAmelCase     =			([prefix + arg for arg in args[0]],)
           lowerCAmelCase     =			True

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





     def 							__UpperCAmelCase      ( self       :      int    ,      UpperCAmelCase__       :      Union[str, Any]    ,      **UpperCAmelCase__       :      str    )				->  Tuple:
        if self.framework == "pt":
           lowerCAmelCase     ,  lowerCAmelCase     =			model_inputs['input_ids'].shape
        elif self.framework == "tf":
           lowerCAmelCase     ,  lowerCAmelCase     =			tf.shape(model_inputs['input_ids']    ).numpy()

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





     def 							__UpperCAmelCase      ( self       :      List[str]    ,      UpperCAmelCase__       :      int    ,      UpperCAmelCase__       :      List[Any]=ReturnType.TEXT    ,      UpperCAmelCase__       :      str=False    )				->  int:
        lowerCAmelCase     =			[]
        for output_ids in model_outputs["output_ids"][0]:
           if return_type == ReturnType.TENSORS:
              lowerCAmelCase     =			{F'''{self.return_name}_token_ids''': output_ids}
           elif return_type == ReturnType.TEXT:
              lowerCAmelCase     =			{
                  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(__lowercase		)
class 		UpperCAmelCase_      (  __lowercase		):
     lowerCamelCase				:	Union[str, Any]		=     '''summary'''





     def __call__( self       :      Any    ,      *UpperCAmelCase__       :      int    ,      **UpperCAmelCase__       :      str    )				->  Union[str, Any]:
        return super().__call__(*UpperCAmelCase__    ,      **UpperCAmelCase__    )





     def 							__UpperCAmelCase      ( self       :      Optional[int]    ,      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(__lowercase		)
class 		UpperCAmelCase_      (  __lowercase		):
     lowerCamelCase				:	Union[str, Any]		=     '''translation'''





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





     def __call__( self       :      Optional[int]    ,      *UpperCAmelCase__       :      int    ,      **UpperCAmelCase__       :      Any    )				->  List[Any]:
        return super().__call__(*UpperCAmelCase__    ,      **UpperCAmelCase__    )










'''simple docstring'''






import gc
import unittest

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

from diffusers import AltDiffusionPipeline, AutoencoderKL, DDIMScheduler, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
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 PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class 				SCREAMING_SNAKE_CASE(							A__							,		A__							,		A__							,		unittest.TestCase			):







	"""simple docstring"""






	lowerCamelCase__			 =       AltDiffusionPipeline
	lowerCamelCase__			 =       TEXT_TO_IMAGE_PARAMS
	lowerCamelCase__			 =       TEXT_TO_IMAGE_BATCH_PARAMS
	lowerCamelCase__			 =       TEXT_TO_IMAGE_IMAGE_PARAMS
	lowerCamelCase__			 =       TEXT_TO_IMAGE_IMAGE_PARAMS







	def      A	(						self			:     Dict		)			->		int:
								torch.manual_seed(0		)
								UpperCAmelCase :       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		,			)
								UpperCAmelCase :       Dict 							=		DDIMScheduler(
								    beta_start=0.0_00_85		,			beta_end=0.0_12		,			beta_schedule='''scaled_linear'''		,			clip_sample=__snake_case		,			set_alpha_to_one=__snake_case		,			)
								torch.manual_seed(0		)
								UpperCAmelCase :       Union[str, Any] 							=		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		,			)

								# TODO: address the non-deterministic text encoder (fails for save-load tests)
								# torch.manual_seed(0)
								# text_encoder_config = RobertaSeriesConfig(
								#     hidden_size=32,
								#     project_dim=32,
								#     intermediate_size=37,
								#     layer_norm_eps=1e-05,
								#     num_attention_heads=4,
								#     num_hidden_layers=5,
								#     vocab_size=5002,
								# )
								# text_encoder = RobertaSeriesModelWithTransformation(text_encoder_config)

								torch.manual_seed(0		)
								UpperCAmelCase :       Dict 							=		CLIPTextConfig(
								    bos_token_id=0		,			eos_token_id=2		,			hidden_size=32		,			projection_dim=32		,			intermediate_size=37		,			layer_norm_eps=1E-05		,			num_attention_heads=4		,			num_hidden_layers=5		,			pad_token_id=1		,			vocab_size=5002		,			)
								UpperCAmelCase :       List[Any] 							=		CLIPTextModel(__snake_case		)

								UpperCAmelCase :       Optional[Any] 							=		XLMRobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-xlm-roberta'''		)
								UpperCAmelCase :       Optional[int] 							=		77

								UpperCAmelCase :       Optional[int] 							=		{
								    '''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]		,			__snake_case			:     Dict		,			__snake_case			:     List[str]=0		)			->		Union[str, Any]:
								if str(__snake_case		).startswith('''mps'''		):
															UpperCAmelCase :       str 							=		torch.manual_seed(__snake_case		)
								else:
															UpperCAmelCase :       Tuple 							=		torch.Generator(device=__snake_case		).manual_seed(__snake_case		)
								UpperCAmelCase :       Dict 							=		{
								    '''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      A	(						self			:     Union[str, Any]		)			->		List[str]:
								super().test_attention_slicing_forward_pass(expected_max_diff=3E-3		)







	def      A	(						self			:     Tuple		)			->		List[str]:
								super().test_inference_batch_single_identical(expected_max_diff=3E-3		)







	def      A	(						self			:     Tuple		)			->		Optional[int]:
								UpperCAmelCase :       List[str] 							=		'''cpu'''  # ensure determinism for the device-dependent torch.Generator

								UpperCAmelCase :       Any 							=		self.get_dummy_components()
								torch.manual_seed(0		)
								UpperCAmelCase :       Optional[int] 							=		RobertaSeriesConfig(
								    hidden_size=32		,			project_dim=32		,			intermediate_size=37		,			layer_norm_eps=1E-05		,			num_attention_heads=4		,			num_hidden_layers=5		,			vocab_size=5002		,			)
								# TODO: remove after fixing the non-deterministic text encoder
								UpperCAmelCase :       List[str] 							=		RobertaSeriesModelWithTransformation(__snake_case		)
								UpperCAmelCase :       str 							=		text_encoder

								UpperCAmelCase :       Optional[int] 							=		AltDiffusionPipeline(**__snake_case		)
								UpperCAmelCase :       str 							=		alt_pipe.to(__snake_case		)
								alt_pipe.set_progress_bar_config(disable=__snake_case		)

								UpperCAmelCase :       Optional[int] 							=		self.get_dummy_inputs(__snake_case		)
								UpperCAmelCase :       Optional[int] 							=		'''A photo of an astronaut'''
								UpperCAmelCase :       List[Any] 							=		alt_pipe(**__snake_case		)
								UpperCAmelCase :       Optional[Any] 							=		output.images
								UpperCAmelCase :       str 							=		image[0, -3:, -3:, -1]

								assert image.shape == (1, 64, 64, 3)
								UpperCAmelCase :       List[str] 							=		np.array(
								    [0.5_74_81_62, 0.60_44_71_45, 0.48_82_12_17, 0.50_10_06_36, 0.5_43_11_85, 0.45_76_36_83, 0.49_65_76_96, 0.48_13_27_33, 0.47_57_30_93]		)

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







	def      A	(						self			:     int		)			->		Union[str, Any]:
								UpperCAmelCase :       Optional[int] 							=		'''cpu'''  # ensure determinism for the device-dependent torch.Generator

								UpperCAmelCase :       int 							=		self.get_dummy_components()
								UpperCAmelCase :       int 							=		PNDMScheduler(skip_prk_steps=__snake_case		)
								torch.manual_seed(0		)
								UpperCAmelCase :       int 							=		RobertaSeriesConfig(
								    hidden_size=32		,			project_dim=32		,			intermediate_size=37		,			layer_norm_eps=1E-05		,			num_attention_heads=4		,			num_hidden_layers=5		,			vocab_size=5002		,			)
								# TODO: remove after fixing the non-deterministic text encoder
								UpperCAmelCase :       Union[str, Any] 							=		RobertaSeriesModelWithTransformation(__snake_case		)
								UpperCAmelCase :       Union[str, Any] 							=		text_encoder
								UpperCAmelCase :       Optional[int] 							=		AltDiffusionPipeline(**__snake_case		)
								UpperCAmelCase :       Dict 							=		alt_pipe.to(__snake_case		)
								alt_pipe.set_progress_bar_config(disable=__snake_case		)

								UpperCAmelCase :       int 							=		self.get_dummy_inputs(__snake_case		)
								UpperCAmelCase :       Optional[int] 							=		alt_pipe(**__snake_case		)
								UpperCAmelCase :       Optional[int] 							=		output.images
								UpperCAmelCase :       int 							=		image[0, -3:, -3:, -1]

								assert image.shape == (1, 64, 64, 3)
								UpperCAmelCase :       Optional[int] 							=		np.array(
								    [0.51_60_50_93, 0.5_70_72_41, 0.47_36_55_07, 0.50_57_88_86, 0.5_63_38_77, 0.4_64_25_03, 0.5_18_20_81, 0.48_76_34_84, 0.49_08_42_37]		)

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

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







	"""simple docstring"""






	def      A	(						self			:     str		)			->		Any:
								# clean up the VRAM after each test
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()







	def      A	(						self			:     List[Any]		)			->		Any:
								# make sure here that pndm scheduler skips prk
								UpperCAmelCase :       List[Any] 							=		AltDiffusionPipeline.from_pretrained('''BAAI/AltDiffusion'''		,			safety_checker=__snake_case		)
								UpperCAmelCase :       Tuple 							=		alt_pipe.to(__snake_case		)
								alt_pipe.set_progress_bar_config(disable=__snake_case		)

								UpperCAmelCase :       List[Any] 							=		'''A painting of a squirrel eating a burger'''
								UpperCAmelCase :       Any 							=		torch.manual_seed(0		)
								UpperCAmelCase :       Optional[int] 							=		alt_pipe([prompt]		,			generator=__snake_case		,			guidance_scale=6.0		,			num_inference_steps=20		,			output_type='''np'''		)

								UpperCAmelCase :       Dict 							=		output.images

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

								assert image.shape == (1, 512, 512, 3)
								UpperCAmelCase :       List[str] 							=		np.array([0.10_10, 0.08_00, 0.07_94, 0.08_85, 0.08_43, 0.07_62, 0.07_69, 0.07_29, 0.05_86]		)

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







	def      A	(						self			:     Tuple		)			->		int:
								UpperCAmelCase :       int 							=		DDIMScheduler.from_pretrained('''BAAI/AltDiffusion'''		,			subfolder='''scheduler'''		)

								UpperCAmelCase :       Tuple 							=		AltDiffusionPipeline.from_pretrained('''BAAI/AltDiffusion'''		,			scheduler=__snake_case		,			safety_checker=__snake_case		)
								UpperCAmelCase :       Dict 							=		alt_pipe.to(__snake_case		)
								alt_pipe.set_progress_bar_config(disable=__snake_case		)

								UpperCAmelCase :       Tuple 							=		'''A painting of a squirrel eating a burger'''
								UpperCAmelCase :       Optional[int] 							=		torch.manual_seed(0		)

								UpperCAmelCase :       List[Any] 							=		alt_pipe([prompt]		,			generator=__snake_case		,			num_inference_steps=2		,			output_type='''numpy'''		)
								UpperCAmelCase :       Dict 							=		output.images

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

								assert image.shape == (1, 512, 512, 3)
								UpperCAmelCase :       Union[str, Any] 							=		np.array([0.40_19, 0.40_52, 0.38_10, 0.41_19, 0.39_16, 0.39_82, 0.46_51, 0.41_95, 0.53_23]		)

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





