import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np



class        lowercase_   :



							def 	UpperCamelCase	(						self    ):
													torch.manual_seed(0    )
													_snake_case						:					Tuple    =   TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					List[Any]    =   AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					Optional[int]    =   UNetaDConditionModel(
													    sample_size=32				,    layers_per_block=1				,    block_out_channels=[32, 64]				,    down_block_types=[
													        "ResnetDownsampleBlock2D",
													        "SimpleCrossAttnDownBlock2D",
													    ]				,    mid_block_type="UNetMidBlock2DSimpleCrossAttn"				,    up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]				,    in_channels=3				,    out_channels=6				,    cross_attention_dim=32				,    encoder_hid_dim=32				,    attention_head_dim=8				,    addition_embed_type="text"				,    addition_embed_type_num_heads=2				,    cross_attention_norm="group_norm"				,    resnet_time_scale_shift="scale_shift"				,    act_fn="gelu"				,    )
													unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													torch.manual_seed(0    )
													_snake_case						:					str    =   DDPMScheduler(
													    num_train_timesteps=1_000				,    beta_schedule="squaredcos_cap_v2"				,    beta_start=0.0_001				,    beta_end=0.02				,    thresholding=lowercase_				,    dynamic_thresholding_ratio=0.95				,    sample_max_value=1.0				,    prediction_type="epsilon"				,    variance_type="learned_range"				,    )

													torch.manual_seed(0    )
													_snake_case						:					int    =   IFWatermarker()

													return {
													    "text_encoder": text_encoder,
													    "tokenizer": tokenizer,
													    "unet": unet,
													    "scheduler": scheduler,
													    "watermarker": watermarker,
													    "safety_checker": None,
													    "feature_extractor": None,
													}



							def 	UpperCamelCase	(						self    ):
													torch.manual_seed(0    )
													_snake_case						:					List[str]    =   TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					Tuple    =   AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					Any    =   UNetaDConditionModel(
													    sample_size=32				,    layers_per_block=[1, 2]				,    block_out_channels=[32, 64]				,    down_block_types=[
													        "ResnetDownsampleBlock2D",
													        "SimpleCrossAttnDownBlock2D",
													    ]				,    mid_block_type="UNetMidBlock2DSimpleCrossAttn"				,    up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]				,    in_channels=6				,    out_channels=6				,    cross_attention_dim=32				,    encoder_hid_dim=32				,    attention_head_dim=8				,    addition_embed_type="text"				,    addition_embed_type_num_heads=2				,    cross_attention_norm="group_norm"				,    resnet_time_scale_shift="scale_shift"				,    act_fn="gelu"				,    class_embed_type="timestep"				,    mid_block_scale_factor=1.414				,    time_embedding_act_fn="gelu"				,    time_embedding_dim=32				,    )
													unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													torch.manual_seed(0    )
													_snake_case						:					Union[str, Any]    =   DDPMScheduler(
													    num_train_timesteps=1_000				,    beta_schedule="squaredcos_cap_v2"				,    beta_start=0.0_001				,    beta_end=0.02				,    thresholding=lowercase_				,    dynamic_thresholding_ratio=0.95				,    sample_max_value=1.0				,    prediction_type="epsilon"				,    variance_type="learned_range"				,    )

													torch.manual_seed(0    )
													_snake_case						:					List[str]    =   DDPMScheduler(
													    num_train_timesteps=1_000				,    beta_schedule="squaredcos_cap_v2"				,    beta_start=0.0_001				,    beta_end=0.02				,    )

													torch.manual_seed(0    )
													_snake_case						:					Dict    =   IFWatermarker()

													return {
													    "text_encoder": text_encoder,
													    "tokenizer": tokenizer,
													    "unet": unet,
													    "scheduler": scheduler,
													    "image_noising_scheduler": image_noising_scheduler,
													    "watermarker": watermarker,
													    "safety_checker": None,
													    "feature_extractor": None,
													}



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					List[Any]    =   self.get_dummy_components()
													_snake_case						:					List[Any]    =   self.pipeline_class(**lowercase_    )
													pipe.to(lowercase_    )
													pipe.set_progress_bar_config(disable=lowercase_    )

													_snake_case						:					Union[str, Any]    =   self.get_dummy_inputs(lowercase_    )

													_snake_case						:					Union[str, Any]    =   inputs["prompt"]
													_snake_case						:					Dict    =   inputs["generator"]
													_snake_case						:					Any    =   inputs["num_inference_steps"]
													_snake_case						:					Union[str, Any]    =   inputs["output_type"]

													if "image" in inputs:
																			_snake_case						:					int    =   inputs["image"]
													else:
																			_snake_case						:					Union[str, Any]    =   None

													if "mask_image" in inputs:
																			_snake_case						:					int    =   inputs["mask_image"]
													else:
																			_snake_case						:					List[str]    =   None

													if "original_image" in inputs:
																			_snake_case						:					Tuple    =   inputs["original_image"]
													else:
																			_snake_case						:					Any    =   None

													_snake_case    ,_snake_case						:					Optional[int]    =   pipe.encode_prompt(lowercase_    )

													# inputs with prompt converted to embeddings
													_snake_case						:					Dict    =   {
													    "prompt_embeds": prompt_embeds,
													    "negative_prompt_embeds": negative_prompt_embeds,
													    "generator": generator,
													    "num_inference_steps": num_inference_steps,
													    "output_type": output_type,
													}

													if image is not None:
																			_snake_case						:					int    =   image

													if mask_image is not None:
																			_snake_case						:					int    =   mask_image

													if original_image is not None:
																			_snake_case						:					Any    =   original_image

													# set all optional components to None
													for optional_component in pipe._optional_components:
																			setattr(lowercase_				,    lowercase_				,    lowercase_    )

													_snake_case						:					Optional[int]    =   pipe(**lowercase_    )[0]

													with tempfile.TemporaryDirectory() as tmpdir:
																			pipe.save_pretrained(lowercase_    )
																			_snake_case						:					Any    =   self.pipeline_class.from_pretrained(lowercase_    )
																			pipe_loaded.to(lowercase_    )
																			pipe_loaded.set_progress_bar_config(disable=lowercase_    )

													pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													for optional_component in pipe._optional_components:
																			self.assertTrue(
																			    getattr(lowercase_				,    lowercase_    ) is None				,    f"""`{optional_component}` did not stay set to None after loading."""				,    )

													_snake_case						:					int    =   self.get_dummy_inputs(lowercase_    )

													_snake_case						:					Optional[int]    =   inputs["generator"]
													_snake_case						:					List[Any]    =   inputs["num_inference_steps"]
													_snake_case						:					Tuple    =   inputs["output_type"]

													# inputs with prompt converted to embeddings
													_snake_case						:					int    =   {
													    "prompt_embeds": prompt_embeds,
													    "negative_prompt_embeds": negative_prompt_embeds,
													    "generator": generator,
													    "num_inference_steps": num_inference_steps,
													    "output_type": output_type,
													}

													if image is not None:
																			_snake_case						:					int    =   image

													if mask_image is not None:
																			_snake_case						:					str    =   mask_image

													if original_image is not None:
																			_snake_case						:					int    =   original_image

													_snake_case						:					Optional[Any]    =   pipe_loaded(**lowercase_    )[0]

													_snake_case						:					Dict    =   np.abs(to_np(lowercase_    ) - to_np(lowercase_    )    ).max()
													self.assertLess(lowercase_				,    1e-4    )



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					Tuple    =   self.get_dummy_components()
													_snake_case						:					Any    =   self.pipeline_class(**lowercase_    )
													pipe.to(lowercase_    )
													pipe.set_progress_bar_config(disable=lowercase_    )

													_snake_case						:					int    =   self.get_dummy_inputs(lowercase_    )
													_snake_case						:					List[str]    =   pipe(**lowercase_    )[0]

													with tempfile.TemporaryDirectory() as tmpdir:
																			pipe.save_pretrained(lowercase_    )
																			_snake_case						:					List[Any]    =   self.pipeline_class.from_pretrained(lowercase_    )
																			pipe_loaded.to(lowercase_    )
																			pipe_loaded.set_progress_bar_config(disable=lowercase_    )

													pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													_snake_case						:					Optional[Any]    =   self.get_dummy_inputs(lowercase_    )
													_snake_case						:					int    =   pipe_loaded(**lowercase_    )[0]

													_snake_case						:					Tuple    =   np.abs(to_np(lowercase_    ) - to_np(lowercase_    )    ).max()
													self.assertLess(lowercase_				,    1e-4    )
import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np



class        lowercase_   :



							def 	UpperCamelCase	(						self    ):
													torch.manual_seed(0    )
													_snake_case						:					Tuple    =   TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					List[Any]    =   AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					Optional[int]    =   UNetaDConditionModel(
													    sample_size=32				,    layers_per_block=1				,    block_out_channels=[32, 64]				,    down_block_types=[
													        "ResnetDownsampleBlock2D",
													        "SimpleCrossAttnDownBlock2D",
													    ]				,    mid_block_type="UNetMidBlock2DSimpleCrossAttn"				,    up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]				,    in_channels=3				,    out_channels=6				,    cross_attention_dim=32				,    encoder_hid_dim=32				,    attention_head_dim=8				,    addition_embed_type="text"				,    addition_embed_type_num_heads=2				,    cross_attention_norm="group_norm"				,    resnet_time_scale_shift="scale_shift"				,    act_fn="gelu"				,    )
													unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													torch.manual_seed(0    )
													_snake_case						:					str    =   DDPMScheduler(
													    num_train_timesteps=1_000				,    beta_schedule="squaredcos_cap_v2"				,    beta_start=0.0_001				,    beta_end=0.02				,    thresholding=lowercase_				,    dynamic_thresholding_ratio=0.95				,    sample_max_value=1.0				,    prediction_type="epsilon"				,    variance_type="learned_range"				,    )

													torch.manual_seed(0    )
													_snake_case						:					int    =   IFWatermarker()

													return {
													    "text_encoder": text_encoder,
													    "tokenizer": tokenizer,
													    "unet": unet,
													    "scheduler": scheduler,
													    "watermarker": watermarker,
													    "safety_checker": None,
													    "feature_extractor": None,
													}



							def 	UpperCamelCase	(						self    ):
													torch.manual_seed(0    )
													_snake_case						:					List[str]    =   TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					Tuple    =   AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					Any    =   UNetaDConditionModel(
													    sample_size=32				,    layers_per_block=[1, 2]				,    block_out_channels=[32, 64]				,    down_block_types=[
													        "ResnetDownsampleBlock2D",
													        "SimpleCrossAttnDownBlock2D",
													    ]				,    mid_block_type="UNetMidBlock2DSimpleCrossAttn"				,    up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]				,    in_channels=6				,    out_channels=6				,    cross_attention_dim=32				,    encoder_hid_dim=32				,    attention_head_dim=8				,    addition_embed_type="text"				,    addition_embed_type_num_heads=2				,    cross_attention_norm="group_norm"				,    resnet_time_scale_shift="scale_shift"				,    act_fn="gelu"				,    class_embed_type="timestep"				,    mid_block_scale_factor=1.414				,    time_embedding_act_fn="gelu"				,    time_embedding_dim=32				,    )
													unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													torch.manual_seed(0    )
													_snake_case						:					Union[str, Any]    =   DDPMScheduler(
													    num_train_timesteps=1_000				,    beta_schedule="squaredcos_cap_v2"				,    beta_start=0.0_001				,    beta_end=0.02				,    thresholding=lowercase_				,    dynamic_thresholding_ratio=0.95				,    sample_max_value=1.0				,    prediction_type="epsilon"				,    variance_type="learned_range"				,    )

													torch.manual_seed(0    )
													_snake_case						:					List[str]    =   DDPMScheduler(
													    num_train_timesteps=1_000				,    beta_schedule="squaredcos_cap_v2"				,    beta_start=0.0_001				,    beta_end=0.02				,    )

													torch.manual_seed(0    )
													_snake_case						:					Dict    =   IFWatermarker()

													return {
													    "text_encoder": text_encoder,
													    "tokenizer": tokenizer,
													    "unet": unet,
													    "scheduler": scheduler,
													    "image_noising_scheduler": image_noising_scheduler,
													    "watermarker": watermarker,
													    "safety_checker": None,
													    "feature_extractor": None,
													}



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					List[Any]    =   self.get_dummy_components()
													_snake_case						:					List[Any]    =   self.pipeline_class(**lowercase_    )
													pipe.to(lowercase_    )
													pipe.set_progress_bar_config(disable=lowercase_    )

													_snake_case						:					Union[str, Any]    =   self.get_dummy_inputs(lowercase_    )

													_snake_case						:					Union[str, Any]    =   inputs["prompt"]
													_snake_case						:					Dict    =   inputs["generator"]
													_snake_case						:					Any    =   inputs["num_inference_steps"]
													_snake_case						:					Union[str, Any]    =   inputs["output_type"]

													if "image" in inputs:
																			_snake_case						:					int    =   inputs["image"]
													else:
																			_snake_case						:					Union[str, Any]    =   None

													if "mask_image" in inputs:
																			_snake_case						:					int    =   inputs["mask_image"]
													else:
																			_snake_case						:					List[str]    =   None

													if "original_image" in inputs:
																			_snake_case						:					Tuple    =   inputs["original_image"]
													else:
																			_snake_case						:					Any    =   None

													_snake_case    ,_snake_case						:					Optional[int]    =   pipe.encode_prompt(lowercase_    )

													# inputs with prompt converted to embeddings
													_snake_case						:					Dict    =   {
													    "prompt_embeds": prompt_embeds,
													    "negative_prompt_embeds": negative_prompt_embeds,
													    "generator": generator,
													    "num_inference_steps": num_inference_steps,
													    "output_type": output_type,
													}

													if image is not None:
																			_snake_case						:					int    =   image

													if mask_image is not None:
																			_snake_case						:					int    =   mask_image

													if original_image is not None:
																			_snake_case						:					Any    =   original_image

													# set all optional components to None
													for optional_component in pipe._optional_components:
																			setattr(lowercase_				,    lowercase_				,    lowercase_    )

													_snake_case						:					Optional[int]    =   pipe(**lowercase_    )[0]

													with tempfile.TemporaryDirectory() as tmpdir:
																			pipe.save_pretrained(lowercase_    )
																			_snake_case						:					Any    =   self.pipeline_class.from_pretrained(lowercase_    )
																			pipe_loaded.to(lowercase_    )
																			pipe_loaded.set_progress_bar_config(disable=lowercase_    )

													pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													for optional_component in pipe._optional_components:
																			self.assertTrue(
																			    getattr(lowercase_				,    lowercase_    ) is None				,    f"""`{optional_component}` did not stay set to None after loading."""				,    )

													_snake_case						:					int    =   self.get_dummy_inputs(lowercase_    )

													_snake_case						:					Optional[int]    =   inputs["generator"]
													_snake_case						:					List[Any]    =   inputs["num_inference_steps"]
													_snake_case						:					Tuple    =   inputs["output_type"]

													# inputs with prompt converted to embeddings
													_snake_case						:					int    =   {
													    "prompt_embeds": prompt_embeds,
													    "negative_prompt_embeds": negative_prompt_embeds,
													    "generator": generator,
													    "num_inference_steps": num_inference_steps,
													    "output_type": output_type,
													}

													if image is not None:
																			_snake_case						:					int    =   image

													if mask_image is not None:
																			_snake_case						:					str    =   mask_image

													if original_image is not None:
																			_snake_case						:					int    =   original_image

													_snake_case						:					Optional[Any]    =   pipe_loaded(**lowercase_    )[0]

													_snake_case						:					Dict    =   np.abs(to_np(lowercase_    ) - to_np(lowercase_    )    ).max()
													self.assertLess(lowercase_				,    1e-4    )



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					Tuple    =   self.get_dummy_components()
													_snake_case						:					Any    =   self.pipeline_class(**lowercase_    )
													pipe.to(lowercase_    )
													pipe.set_progress_bar_config(disable=lowercase_    )

													_snake_case						:					int    =   self.get_dummy_inputs(lowercase_    )
													_snake_case						:					List[str]    =   pipe(**lowercase_    )[0]

													with tempfile.TemporaryDirectory() as tmpdir:
																			pipe.save_pretrained(lowercase_    )
																			_snake_case						:					List[Any]    =   self.pipeline_class.from_pretrained(lowercase_    )
																			pipe_loaded.to(lowercase_    )
																			pipe_loaded.set_progress_bar_config(disable=lowercase_    )

													pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													_snake_case						:					Optional[Any]    =   self.get_dummy_inputs(lowercase_    )
													_snake_case						:					int    =   pipe_loaded(**lowercase_    )[0]

													_snake_case						:					Tuple    =   np.abs(to_np(lowercase_    ) - to_np(lowercase_    )    ).max()
													self.assertLess(lowercase_				,    1e-4    )