






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 							snake_case_       (    __UpperCamelCase		,							__UpperCamelCase		,							__UpperCamelCase		,							unittest.TestCase      ):

						"""simple docstring"""



						snake_case__        =      AltDiffusionPipeline
						snake_case__        =      TEXT_TO_IMAGE_PARAMS
						snake_case__        =      TEXT_TO_IMAGE_BATCH_PARAMS
						snake_case__        =      TEXT_TO_IMAGE_IMAGE_PARAMS
						snake_case__        =      TEXT_TO_IMAGE_IMAGE_PARAMS



						def        UpperCAmelCase__					(self:					Any   )    ->     List[Any]:




											'''simple docstring'''



											torch.manual_seed(0   )
											__a					:						Union[str, Any]      =					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 ,					)
											__a					:						Tuple      =					DDIMScheduler(
											    beta_start=0.0_00_85 ,					beta_end=0.0_12 ,					beta_schedule="scaled_linear" ,					clip_sample=__UpperCAmelCase ,					set_alpha_to_one=__UpperCAmelCase ,					)
											torch.manual_seed(0   )
											__a					:						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 ,					)

											# 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   )
											__a					:						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 ,					)
											__a					:						List[str]      =					CLIPTextModel(__UpperCAmelCase   )

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

											__a					:						List[str]      =					{
											    "unet": unet,
											    "scheduler": scheduler,
											    "vae": vae,
											    "text_encoder": text_encoder,
											    "tokenizer": tokenizer,
											    "safety_checker": None,
											    "feature_extractor": None,
											}
											return components



						def        UpperCAmelCase__					(self:					Optional[Any] ,					__UpperCAmelCase:					int ,					__UpperCAmelCase:					str=0   )    ->     Any:




											'''simple docstring'''



											if str(__UpperCAmelCase   ).startswith("mps"   ):
																__a					:						Optional[int]      =					torch.manual_seed(__UpperCAmelCase   )
											else:
																__a					:						Any      =					torch.Generator(device=__UpperCAmelCase   ).manual_seed(__UpperCAmelCase   )
											__a					:						int      =					{
											    "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:					List[str]   )    ->     str:




											'''simple docstring'''



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



						def        UpperCAmelCase__					(self:					List[str]   )    ->     Dict:




											'''simple docstring'''



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



						def        UpperCAmelCase__					(self:					Tuple   )    ->     List[str]:




											'''simple docstring'''



											__a					:						Optional[int]      =					"cpu"  # ensure determinism for the device-dependent torch.Generator

											__a					:						int      =					self.get_dummy_components()
											torch.manual_seed(0   )
											__a					:						List[Any]      =					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
											__a					:						Dict      =					RobertaSeriesModelWithTransformation(__UpperCAmelCase   )
											__a					:						Optional[Any]      =					text_encoder

											__a					:						Any      =					AltDiffusionPipeline(**__UpperCAmelCase   )
											__a					:						str      =					alt_pipe.to(__UpperCAmelCase   )
											alt_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

											__a					:						Any      =					self.get_dummy_inputs(__UpperCAmelCase   )
											__a					:						Union[str, Any]      =					"A photo of an astronaut"
											__a					:						Dict      =					alt_pipe(**__UpperCAmelCase   )
											__a					:						Any      =					output.images
											__a					:						int      =					image[0, -3:, -3:, -1]

											assert image.shape == (1, 64, 64, 3)
											__a					:						Union[str, Any]      =					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        UpperCAmelCase__					(self:					Union[str, Any]   )    ->     Dict:




											'''simple docstring'''



											__a					:						Union[str, Any]      =					"cpu"  # ensure determinism for the device-dependent torch.Generator

											__a					:						str      =					self.get_dummy_components()
											__a					:						str      =					PNDMScheduler(skip_prk_steps=__UpperCAmelCase   )
											torch.manual_seed(0   )
											__a					:						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
											__a					:						Optional[Any]      =					RobertaSeriesModelWithTransformation(__UpperCAmelCase   )
											__a					:						Optional[int]      =					text_encoder
											__a					:						int      =					AltDiffusionPipeline(**__UpperCAmelCase   )
											__a					:						List[str]      =					alt_pipe.to(__UpperCAmelCase   )
											alt_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

											__a					:						Union[str, Any]      =					self.get_dummy_inputs(__UpperCAmelCase   )
											__a					:						Optional[int]      =					alt_pipe(**__UpperCAmelCase   )
											__a					:						str      =					output.images
											__a					:						List[str]      =					image[0, -3:, -3:, -1]

											assert image.shape == (1, 64, 64, 3)
											__a					:						Union[str, Any]      =					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 							snake_case_       (    unittest.TestCase      ):

						"""simple docstring"""



						def        UpperCAmelCase__					(self:					Dict   )    ->     List[str]:




											'''simple docstring'''



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



						def        UpperCAmelCase__					(self:					int   )    ->     Tuple:




											'''simple docstring'''



											__a					:						Union[str, Any]      =					AltDiffusionPipeline.from_pretrained("BAAI/AltDiffusion" ,					safety_checker=__UpperCAmelCase   )
											__a					:						str      =					alt_pipe.to(__UpperCAmelCase   )
											alt_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

											__a					:						List[str]      =					"A painting of a squirrel eating a burger"
											__a					:						List[Any]      =					torch.manual_seed(0   )
											__a					:						Any      =					alt_pipe([prompt] ,					generator=__UpperCAmelCase ,					guidance_scale=6.0 ,					num_inference_steps=20 ,					output_type="np"   )

											__a					:						Tuple      =					output.images

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

											assert image.shape == (1, 512, 512, 3)
											__a					:						Tuple      =					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        UpperCAmelCase__					(self:					str   )    ->     Tuple:




											'''simple docstring'''



											__a					:						Optional[int]      =					DDIMScheduler.from_pretrained("BAAI/AltDiffusion" ,					subfolder="scheduler"   )

											__a					:						Dict      =					AltDiffusionPipeline.from_pretrained("BAAI/AltDiffusion" ,					scheduler=__UpperCAmelCase ,					safety_checker=__UpperCAmelCase   )
											__a					:						int      =					alt_pipe.to(__UpperCAmelCase   )
											alt_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

											__a					:						str      =					"A painting of a squirrel eating a burger"
											__a					:						Optional[int]      =					torch.manual_seed(0   )

											__a					:						List[str]      =					alt_pipe([prompt] ,					generator=__UpperCAmelCase ,					num_inference_steps=2 ,					output_type="numpy"   )
											__a					:						Tuple      =					output.images

											__a					:						Any      =					image[0, -3:, -3:, -1]

											assert image.shape == (1, 512, 512, 3)
											__a					:						List[str]      =					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













import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
			import torch_xla.core.xla_model as xm
			import torch_xla.debug.metrics as met




class 							snake_case_       (    __UpperCamelCase      ):

						"""simple docstring"""



						def __init__(self:					str ,					*__UpperCAmelCase:					Optional[int] ,					__UpperCAmelCase:					Optional[int]=None ,					__UpperCAmelCase:					Dict=None ,					**__UpperCAmelCase:					Tuple   )    ->     Dict:




											'''simple docstring'''



											super().__init__(*__UpperCAmelCase ,					**__UpperCAmelCase   )
											__a					:						str      =					eval_examples
											__a					:						List[Any]      =					post_process_function



						def        UpperCAmelCase__					(self:					Optional[int] ,					__UpperCAmelCase:					Optional[Dataset] = None ,					__UpperCAmelCase:					Optional[int]=None ,					__UpperCAmelCase:					Optional[List[str]] = None ,					__UpperCAmelCase:					str = "eval" ,					**__UpperCAmelCase:					Any ,					)    ->     Dict[str, float]:




											'''simple docstring'''



											__a					:						Optional[int]      =					gen_kwargs.copy()
											__a					:						Union[str, Any]      =					(
											    gen_kwargs["max_length"] if gen_kwargs.get("max_length"   ) is not None else self.args.generation_max_length
											)
											__a					:						Dict      =					(
											    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"   ) is not None else self.args.generation_num_beams
											)
											__a					:						Optional[Any]      =					gen_kwargs

											__a					:						Union[str, Any]      =					self.eval_dataset if eval_dataset is None else eval_dataset
											__a					:						Optional[int]      =					self.get_eval_dataloader(__UpperCAmelCase   )
											__a					:						Tuple      =					self.eval_examples if eval_examples is None else eval_examples

											# Temporarily disable metric computation, we will do it in the loop here.
											__a					:						int      =					self.compute_metrics
											__a					:						Dict      =					None
											__a					:						List[Any]      =					time.time()
											__a					:						str      =					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
											try:
																__a					:						str      =					eval_loop(
																    __UpperCAmelCase ,					description="Evaluation" ,					prediction_loss_only=True if compute_metrics is None else None ,					ignore_keys=__UpperCAmelCase ,					metric_key_prefix=__UpperCAmelCase ,					)
											finally:
																__a					:						Union[str, Any]      =					compute_metrics
											__a					:						str      =					self.args.eval_batch_size * self.args.world_size
											if f'{metric_key_prefix}_jit_compilation_time' in output.metrics:
																start_time += output.metrics[f'{metric_key_prefix}_jit_compilation_time']
											output.metrics.update(
											    speed_metrics(
											        __UpperCAmelCase ,					__UpperCAmelCase ,					num_samples=output.num_samples ,					num_steps=math.ceil(output.num_samples / total_batch_size   ) ,					)   )

											if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
																# Only the main node write the results by default
																__a					:						Union[str, Any]      =					self.post_process_function(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase   )
																__a					:						Dict      =					self.compute_metrics(__UpperCAmelCase   )

																# Prefix all keys with metric_key_prefix + '_'
																for key in list(metrics.keys()   ):
																					if not key.startswith(f'{metric_key_prefix}_'   ):
																										__a					:						int      =					metrics.pop(__UpperCAmelCase   )

																metrics.update(output.metrics   )
											else:
																__a					:						Optional[int]      =					output.metrics

											if self.args.should_log:
																# Only the main node log the results by default
																self.log(__UpperCAmelCase   )

											if self.args.tpu_metrics_debug or self.args.debug:
																# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
																xm.master_print(met.metrics_report()   )

											__a					:						Tuple      =					self.callback_handler.on_evaluate(self.args ,					self.state ,					self.control ,					__UpperCAmelCase   )
											return metrics



						def        UpperCAmelCase__					(self:					Dict ,					__UpperCAmelCase:					List[str] ,					__UpperCAmelCase:					List[str] ,					__UpperCAmelCase:					Union[str, Any]=None ,					__UpperCAmelCase:					str = "test" ,					**__UpperCAmelCase:					List[Any]   )    ->     Union[str, Any]:




											'''simple docstring'''



											__a					:						Any      =					gen_kwargs.copy()

											__a					:						Tuple      =					self.get_test_dataloader(__UpperCAmelCase   )

											# Temporarily disable metric computation, we will do it in the loop here.
											__a					:						Optional[Any]      =					self.compute_metrics
											__a					:						int      =					None
											__a					:						List[Any]      =					time.time()
											__a					:						Dict      =					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
											try:
																__a					:						Optional[Any]      =					eval_loop(
																    __UpperCAmelCase ,					description="Prediction" ,					prediction_loss_only=True if compute_metrics is None else None ,					ignore_keys=__UpperCAmelCase ,					metric_key_prefix=__UpperCAmelCase ,					)
											finally:
																__a					:						List[Any]      =					compute_metrics

											__a					:						Optional[Any]      =					self.args.eval_batch_size * self.args.world_size
											if f'{metric_key_prefix}_jit_compilation_time' in output.metrics:
																start_time += output.metrics[f'{metric_key_prefix}_jit_compilation_time']
											output.metrics.update(
											    speed_metrics(
											        __UpperCAmelCase ,					__UpperCAmelCase ,					num_samples=output.num_samples ,					num_steps=math.ceil(output.num_samples / total_batch_size   ) ,					)   )
											if self.post_process_function is None or self.compute_metrics is None:
																return output

											__a					:						Optional[int]      =					self.post_process_function(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					"predict"   )
											__a					:						Dict      =					self.compute_metrics(__UpperCAmelCase   )

											# Prefix all keys with metric_key_prefix + '_'
											for key in list(metrics.keys()   ):
																if not key.startswith(f'{metric_key_prefix}_'   ):
																					__a					:						List[str]      =					metrics.pop(__UpperCAmelCase   )
											metrics.update(output.metrics   )
											return PredictionOutput(predictions=predictions.predictions ,					label_ids=predictions.label_ids ,					metrics=__UpperCAmelCase   )





