




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       _A				(		__lowercase       ,     __lowercase       ,     __lowercase       ,     unittest.TestCase     ):
		__a					    =       AltDiffusionPipeline
		__a					    =       TEXT_TO_IMAGE_PARAMS
		__a					    =       TEXT_TO_IMAGE_BATCH_PARAMS
		__a					    =       TEXT_TO_IMAGE_IMAGE_PARAMS
		__a					    =       TEXT_TO_IMAGE_IMAGE_PARAMS






		def 				UpperCAmelCase (    self			):
								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.0_0085						,		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						,		)

								# 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      =			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      =			CLIPTextModel(__lowerCAmelCase			)

								_UpperCAmelCase      =			XLMRobertaTokenizer.from_pretrained("""hf-internal-testing/tiny-xlm-roberta"""			)
								_UpperCAmelCase      =			77

								_UpperCAmelCase      =			{
								    """unet""": unet,
								    """scheduler""": scheduler,
								    """vae""": vae,
								    """text_encoder""": text_encoder,
								    """tokenizer""": tokenizer,
								    """safety_checker""": None,
								    """feature_extractor""": None,
								}
								return components






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=0			):







								if str(__lowerCAmelCase			).startswith("""mps"""			):
														_UpperCAmelCase      =			torch.manual_seed(__lowerCAmelCase			)
								else:
														_UpperCAmelCase      =			torch.Generator(device=__lowerCAmelCase			).manual_seed(__lowerCAmelCase			)
								_UpperCAmelCase      =			{
								    """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 				UpperCAmelCase (    self			):
								super().test_attention_slicing_forward_pass(expected_max_diff=3e-3			)






		def 				UpperCAmelCase (    self			):
								super().test_inference_batch_single_identical(expected_max_diff=3e-3			)






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator

								_UpperCAmelCase      =			self.get_dummy_components()
								torch.manual_seed(0			)
								_UpperCAmelCase      =			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      =			RobertaSeriesModelWithTransformation(__lowerCAmelCase			)
								_UpperCAmelCase      =			text_encoder

								_UpperCAmelCase      =			AltDiffusionPipeline(**__lowerCAmelCase			)
								_UpperCAmelCase      =			alt_pipe.to(__lowerCAmelCase			)
								alt_pipe.set_progress_bar_config(disable=__lowerCAmelCase			)

								_UpperCAmelCase      =			self.get_dummy_inputs(__lowerCAmelCase			)
								_UpperCAmelCase      =			"""A photo of an astronaut"""
								_UpperCAmelCase      =			alt_pipe(**__lowerCAmelCase			)
								_UpperCAmelCase      =			output.images
								_UpperCAmelCase      =			image[0, -3:, -3:, -1]

								assert image.shape == (1, 64, 64, 3)
								_UpperCAmelCase      =			np.array(
								    [0.574_8162, 0.6044_7145, 0.4882_1217, 0.5010_0636, 0.543_1185, 0.4576_3683, 0.4965_7696, 0.4813_2733, 0.4757_3093]			)

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






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator

								_UpperCAmelCase      =			self.get_dummy_components()
								_UpperCAmelCase      =			PNDMScheduler(skip_prk_steps=__lowerCAmelCase			)
								torch.manual_seed(0			)
								_UpperCAmelCase      =			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      =			RobertaSeriesModelWithTransformation(__lowerCAmelCase			)
								_UpperCAmelCase      =			text_encoder
								_UpperCAmelCase      =			AltDiffusionPipeline(**__lowerCAmelCase			)
								_UpperCAmelCase      =			alt_pipe.to(__lowerCAmelCase			)
								alt_pipe.set_progress_bar_config(disable=__lowerCAmelCase			)

								_UpperCAmelCase      =			self.get_dummy_inputs(__lowerCAmelCase			)
								_UpperCAmelCase      =			alt_pipe(**__lowerCAmelCase			)
								_UpperCAmelCase      =			output.images
								_UpperCAmelCase      =			image[0, -3:, -3:, -1]

								assert image.shape == (1, 64, 64, 3)
								_UpperCAmelCase      =			np.array(
								    [0.5160_5093, 0.570_7241, 0.4736_5507, 0.5057_8886, 0.563_3877, 0.464_2503, 0.518_2081, 0.4876_3484, 0.4908_4237]			)

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




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






		def 				UpperCAmelCase (    self			):
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			AltDiffusionPipeline.from_pretrained("""BAAI/AltDiffusion"""						,		safety_checker=__lowerCAmelCase			)
								_UpperCAmelCase      =			alt_pipe.to(__lowerCAmelCase			)
								alt_pipe.set_progress_bar_config(disable=__lowerCAmelCase			)

								_UpperCAmelCase      =			"""A painting of a squirrel eating a burger"""
								_UpperCAmelCase      =			torch.manual_seed(0			)
								_UpperCAmelCase      =			alt_pipe([prompt]						,		generator=__lowerCAmelCase						,		guidance_scale=6.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.1010, 0.0800, 0.0794, 0.0885, 0.0843, 0.0762, 0.0769, 0.0729, 0.0586]			)

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






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			DDIMScheduler.from_pretrained("""BAAI/AltDiffusion"""						,		subfolder="""scheduler"""			)

								_UpperCAmelCase      =			AltDiffusionPipeline.from_pretrained("""BAAI/AltDiffusion"""						,		scheduler=__lowerCAmelCase						,		safety_checker=__lowerCAmelCase			)
								_UpperCAmelCase      =			alt_pipe.to(__lowerCAmelCase			)
								alt_pipe.set_progress_bar_config(disable=__lowerCAmelCase			)

								_UpperCAmelCase      =			"""A painting of a squirrel eating a burger"""
								_UpperCAmelCase      =			torch.manual_seed(0			)

								_UpperCAmelCase      =			alt_pipe([prompt]						,		generator=__lowerCAmelCase						,		num_inference_steps=2						,		output_type="""numpy"""			)
								_UpperCAmelCase      =			output.images

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

								assert image.shape == (1, 512, 512, 3)
								_UpperCAmelCase      =			np.array([0.4019, 0.4052, 0.3810, 0.4119, 0.3916, 0.3982, 0.4651, 0.4195, 0.5323]			)

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





import uuid
from typing import Any, Dict, List, Optional, Union

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

if is_torch_available():
				import torch


a													=			logging.get_logger(__name__)




class       _A				:






		def __init__(    self						,		_SCREAMING_SNAKE_CASE = None						,		_SCREAMING_SNAKE_CASE = None						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None			):
								if not conversation_id:
														_UpperCAmelCase      =			uuid.uuida()
								if past_user_inputs is None:
														_UpperCAmelCase      =			[]
								if generated_responses is None:
														_UpperCAmelCase      =			[]

								_UpperCAmelCase      =			conversation_id
								_UpperCAmelCase      =			past_user_inputs
								_UpperCAmelCase      =			generated_responses
								_UpperCAmelCase      =			text






		def __eq__(    self						,		_SCREAMING_SNAKE_CASE			):
								if not isinstance(_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE			):
														return False
								if self.uuid == other.uuid:
														return True
								return (
								    self.new_user_input == other.new_user_input
								    and self.past_user_inputs == other.past_user_inputs
								    and self.generated_responses == other.generated_responses
								)






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE = False			):
								if self.new_user_input:
														if overwrite:
																				logger.warning(
																				    F"User input added while unprocessed input was existing: \"{self.new_user_input}\" was overwritten "
																				    F"with: \"{text}\"."			)
																				_UpperCAmelCase      =			text
														else:
																				logger.warning(
																				    F"User input added while unprocessed input was existing: \"{self.new_user_input}\" new input "
																				    F"ignored: \"{text}\". Set `overwrite` to True to overwrite unprocessed user input"			)
								else:
														_UpperCAmelCase      =			text






		def 				UpperCAmelCase (    self			):
								if self.new_user_input:
														self.past_user_inputs.append(self.new_user_input			)
								_UpperCAmelCase      =			None






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								self.generated_responses.append(_SCREAMING_SNAKE_CASE			)






		def 				UpperCAmelCase (    self			):
								for user_input, generated_response in zip(self.past_user_inputs						,		self.generated_responses			):
														yield True, user_input
														yield False, generated_response
								if self.new_user_input:
														yield True, self.new_user_input






		def __repr__(    self			):
								_UpperCAmelCase      =			F"Conversation id: {self.uuid} \n"
								for is_user, text in self.iter_texts():
														_UpperCAmelCase      =			"""user""" if is_user else """bot"""
														output += F"{name} >> {text} \n"
								return output




@add_end_docstrings(
    __lowercase       ,     R"""
        min_length_for_response (`int`, *optional*, defaults to 32):
            The minimum length (in number of tokens) for a response.
        minimum_tokens (`int`, *optional*, defaults to 10):
            The minimum length of tokens to leave for a response.
    """       ,     )
class       _A				(		__lowercase     ):






		def __init__(    self						,		*_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			):
								super().__init__(*_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			)
								if self.tokenizer.pad_token_id is None:
														_UpperCAmelCase      =			self.tokenizer.eos_token






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None						,		**_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			{}
								_UpperCAmelCase      =			{}
								_UpperCAmelCase      =			{}

								if min_length_for_response is not None:
														_UpperCAmelCase      =			min_length_for_response
								if minimum_tokens is not None:
														_UpperCAmelCase      =			minimum_tokens

								if "max_length" in generate_kwargs:
														_UpperCAmelCase      =			generate_kwargs["""max_length"""]
														# self.max_length = generate_kwargs.get("max_length", self.model.config.max_length)
								if clean_up_tokenization_spaces is not None:
														_UpperCAmelCase      =			clean_up_tokenization_spaces

								if generate_kwargs:
														forward_params.update(_SCREAMING_SNAKE_CASE			)
								return preprocess_params, forward_params, postprocess_params






		def __call__(    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=0						,		**_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			super().__call__(_SCREAMING_SNAKE_CASE						,		num_workers=_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			)
								if isinstance(_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE			) and len(_SCREAMING_SNAKE_CASE			) == 1:
														return outputs[0]
								return outputs






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=32			):
								if not isinstance(_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE			):
														raise ValueError("""ConversationalPipeline, expects Conversation as inputs"""			)
								if conversation.new_user_input is None:
														raise ValueError(
														    F"Conversation with UUID {type(conversation.uuid			)} does not contain new user input to process. "
														    """Add user inputs with the conversation's `add_user_input` method"""			)
								if hasattr(self.tokenizer						,		"""_build_conversation_input_ids"""			):
														_UpperCAmelCase      =			self.tokenizer._build_conversation_input_ids(_SCREAMING_SNAKE_CASE			)
								else:
														# If the tokenizer cannot handle conversations, we default to only the old version
														_UpperCAmelCase      =			self._legacy_parse_and_tokenize(_SCREAMING_SNAKE_CASE			)

								if self.framework == "pt":
														_UpperCAmelCase      =			torch.LongTensor([input_ids]			)
								elif self.framework == "tf":
														_UpperCAmelCase      =			tf.constant([input_ids]			)
								return {"input_ids": input_ids, "conversation": conversation}






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=10						,		**_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			generate_kwargs.get("""max_length"""						,		self.model.config.max_length			)

								_UpperCAmelCase      =			model_inputs["""input_ids"""].shape[1]
								if max_length - minimum_tokens < n:
														logger.warning(F"Conversation input is to long ({n}), trimming it to ({max_length} - {minimum_tokens})"			)
														_UpperCAmelCase      =			max_length - minimum_tokens
														_UpperCAmelCase      =			model_inputs["""input_ids"""][:, -trim:]
														if "attention_mask" in model_inputs:
																				_UpperCAmelCase      =			model_inputs["""attention_mask"""][:, -trim:]
								_UpperCAmelCase      =			model_inputs.pop("""conversation"""			)
								_UpperCAmelCase      =			max_length
								_UpperCAmelCase      =			self.model.generate(**_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			)
								if self.model.config.is_encoder_decoder:
														_UpperCAmelCase      =			1
								else:
														_UpperCAmelCase      =			n
								return {"output_ids": output_ids[:, start_position:], "conversation": conversation}






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=True			):
								_UpperCAmelCase      =			model_outputs["""output_ids"""]
								_UpperCAmelCase      =			self.tokenizer.decode(
								    output_ids[0]						,		skip_special_tokens=_SCREAMING_SNAKE_CASE						,		clean_up_tokenization_spaces=_SCREAMING_SNAKE_CASE						,		)
								_UpperCAmelCase      =			model_outputs["""conversation"""]
								conversation.mark_processed()
								conversation.append_response(_SCREAMING_SNAKE_CASE			)
								return conversation






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			self.tokenizer.eos_token_id
								_UpperCAmelCase      =			[]
								for is_user, text in conversation.iter_texts():
														if eos_token_id is not None:
																				input_ids.extend(self.tokenizer.encode(_SCREAMING_SNAKE_CASE						,		add_special_tokens=_SCREAMING_SNAKE_CASE			) + [eos_token_id]			)
														else:
																				input_ids.extend(self.tokenizer.encode(_SCREAMING_SNAKE_CASE						,		add_special_tokens=_SCREAMING_SNAKE_CASE			)			)

								if len(_SCREAMING_SNAKE_CASE			) > self.tokenizer.model_max_length:
														_UpperCAmelCase      =			input_ids[-self.tokenizer.model_max_length :]
								return input_ids