






'''simple docstring'''




import inspect
import os

import torch
from transformers import AutoModel
from transformers.testing_utils import mockenv_context
from transformers.trainer_utils import set_seed

import accelerate
from accelerate.accelerator import Accelerator
from accelerate.state import AcceleratorState
from accelerate.test_utils.testing import (
    AccelerateTestCase,
    TempDirTestCase,
    execute_subprocess_async,
    require_cuda,
    require_fsdp,
    require_multi_gpu,
    slow,
)
from accelerate.utils.constants import (
    FSDP_AUTO_WRAP_POLICY,
    FSDP_BACKWARD_PREFETCH,
    FSDP_SHARDING_STRATEGY,
    FSDP_STATE_DICT_TYPE,
)
from accelerate.utils.dataclasses import FullyShardedDataParallelPlugin
from accelerate.utils.other import patch_environment


set_seed(42)

lowerCAmelCase__          =						'bert-base-cased'
lowerCAmelCase__          =						'fp16'
lowerCAmelCase__          =						'bf16'
lowerCAmelCase__          =						[FPaa, BFaa]







@require_fsdp
@require_cuda
class       __lowercase					(__lowerCamelCase    ):

	def      __UpperCamelCase							(      self   :						Dict):
			super().setUp()

			UpperCamelCase__				:					str					=					dict(
			    ACCELERATE_USE_FSDP='true'  ,					MASTER_ADDR='localhost'  ,					MASTER_PORT='10999'  ,					RANK='0'  ,					LOCAL_RANK='0'  ,					WORLD_SIZE='1'  ,					)

	def      __UpperCamelCase							(      self   :						Dict):
			from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

			for i, strategy in enumerate(UpperCAmelCase_):
					UpperCamelCase__				:					Any					=					self.dist_env.copy()
					UpperCamelCase__				:					Optional[int]					=					F'{i + 1}'
					UpperCamelCase__				:					List[str]					=					strategy
					with mockenv_context(**UpperCAmelCase_):
							UpperCamelCase__				:					Optional[Any]					=					FullyShardedDataParallelPlugin()
							self.assertEqual(fsdp_plugin.sharding_strategy  ,					ShardingStrategy(i + 1))

	def      __UpperCamelCase							(      self   :						Optional[int]):
			from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

			for i, prefetch_policy in enumerate(UpperCAmelCase_):
					UpperCamelCase__				:					Tuple					=					self.dist_env.copy()
					UpperCamelCase__				:					int					=					prefetch_policy
					with mockenv_context(**UpperCAmelCase_):
							UpperCamelCase__				:					Optional[Any]					=					FullyShardedDataParallelPlugin()
							if prefetch_policy == "NO_PREFETCH":
									self.assertIsNone(fsdp_plugin.backward_prefetch)
							else:
									self.assertEqual(fsdp_plugin.backward_prefetch  ,					BackwardPrefetch(i + 1))

	def      __UpperCamelCase							(      self   :						Any):
			from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

			for i, state_dict_type in enumerate(UpperCAmelCase_):
					UpperCamelCase__				:					List[str]					=					self.dist_env.copy()
					UpperCamelCase__				:					Dict					=					state_dict_type
					with mockenv_context(**UpperCAmelCase_):
							UpperCamelCase__				:					Any					=					FullyShardedDataParallelPlugin()
							self.assertEqual(fsdp_plugin.state_dict_type  ,					StateDictType(i + 1))
							if state_dict_type == "FULL_STATE_DICT":
									self.assertTrue(fsdp_plugin.state_dict_config.offload_to_cpu)
									self.assertTrue(fsdp_plugin.state_dict_config.ranka_only)

	def      __UpperCamelCase							(      self   :						Union[str, Any]):
			UpperCamelCase__				:					Union[str, Any]					=					AutoModel.from_pretrained(UpperCAmelCase_)
			for policy in FSDP_AUTO_WRAP_POLICY:
					UpperCamelCase__				:					Optional[int]					=					self.dist_env.copy()
					UpperCamelCase__				:					List[str]					=					policy
					if policy == "TRANSFORMER_BASED_WRAP":
							UpperCamelCase__				:					List[Any]					=					'BertLayer'
					elif policy == "SIZE_BASED_WRAP":
							UpperCamelCase__				:					int					=					'2000'
					with mockenv_context(**UpperCAmelCase_):
							UpperCamelCase__				:					int					=					FullyShardedDataParallelPlugin()
							fsdp_plugin.set_auto_wrap_policy(UpperCAmelCase_)
							if policy == "NO_WRAP":
									self.assertIsNone(fsdp_plugin.auto_wrap_policy)
							else:
									self.assertIsNotNone(fsdp_plugin.auto_wrap_policy)

			UpperCamelCase__				:					int					=					self.dist_env.copy()
			UpperCamelCase__				:					List[str]					=					'TRANSFORMER_BASED_WRAP'
			UpperCamelCase__				:					List[str]					=					'T5Layer'
			with mockenv_context(**UpperCAmelCase_):
					UpperCamelCase__				:					Optional[Any]					=					FullyShardedDataParallelPlugin()
					with self.assertRaises(UpperCAmelCase_) as cm:
							fsdp_plugin.set_auto_wrap_policy(UpperCAmelCase_)
					self.assertTrue('Could not find the transformer layer class to wrap in the model.' in str(cm.exception))

			UpperCamelCase__				:					Optional[Any]					=					self.dist_env.copy()
			UpperCamelCase__				:					int					=					'SIZE_BASED_WRAP'
			UpperCamelCase__				:					Optional[int]					=					'0'
			with mockenv_context(**UpperCAmelCase_):
					UpperCamelCase__				:					Tuple					=					FullyShardedDataParallelPlugin()
					fsdp_plugin.set_auto_wrap_policy(UpperCAmelCase_)
					self.assertIsNone(fsdp_plugin.auto_wrap_policy)

	def      __UpperCamelCase							(      self   :						Any):
			from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
			from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

			for mp_dtype in dtypes:
					UpperCamelCase__				:					Union[str, Any]					=					self.dist_env.copy()
					UpperCamelCase__				:					Optional[int]					=					mp_dtype
					with mockenv_context(**UpperCAmelCase_):
							UpperCamelCase__				:					int					=					Accelerator()
							if mp_dtype == "fp16":
									UpperCamelCase__				:					Optional[Any]					=					torch.floataa
							elif mp_dtype == "bf16":
									UpperCamelCase__				:					int					=					torch.bfloataa
							UpperCamelCase__				:					List[str]					=					MixedPrecision(param_dtype=UpperCAmelCase_  ,					reduce_dtype=UpperCAmelCase_  ,					buffer_dtype=UpperCAmelCase_)
							self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy  ,					UpperCAmelCase_)
							if mp_dtype == FPaa:
									self.assertTrue(isinstance(accelerator.scaler  ,					UpperCAmelCase_))
							elif mp_dtype == BFaa:
									self.assertIsNone(accelerator.scaler)
							AcceleratorState._reset_state(UpperCAmelCase_)





	def      __UpperCamelCase							(      self   :						Optional[Any]):
			from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

			for flag in [True, False]:
					UpperCamelCase__				:					str					=					self.dist_env.copy()
					UpperCamelCase__				:					List[str]					=					str(UpperCAmelCase_).lower()
					with mockenv_context(**UpperCAmelCase_):
							UpperCamelCase__				:					str					=					FullyShardedDataParallelPlugin()
							self.assertEqual(fsdp_plugin.cpu_offload  ,					CPUOffload(offload_params=UpperCAmelCase_))







@require_fsdp
@require_multi_gpu
@slow
class       __lowercase					(__lowerCamelCase    ):

	def      __UpperCamelCase							(      self   :						Dict):
			super().setUp()
			UpperCamelCase__				:					List[Any]					=					0.82
			UpperCamelCase__				:					Optional[Any]					=					[
			    'fsdp_shard_grad_op_transformer_based_wrap',
			    'fsdp_full_shard_transformer_based_wrap',
			]
			UpperCamelCase__				:					List[Any]					=					{
			    'multi_gpu_fp16': 3_200,
			    'fsdp_shard_grad_op_transformer_based_wrap_fp16': 2_000,
			    'fsdp_full_shard_transformer_based_wrap_fp16': 1_900,
			    # Disabling below test as it overwhelms the RAM memory usage
			    # on CI self-hosted runner leading to tests getting killed.
			    # "fsdp_full_shard_cpu_offload_transformer_based_wrap_fp32": 1500,  # fp16 was leading to indefinite hang
			}
			UpperCamelCase__				:					Tuple					=					160
			UpperCamelCase__				:					str					=					160

			UpperCamelCase__				:					List[Any]					=					inspect.getfile(accelerate.test_utils)
			UpperCamelCase__				:					int					=					os.path.sep.join(mod_file.split(os.path.sep)[:-1] + ['scripts', 'external_deps'])

	def      __UpperCamelCase							(      self   :						List[str]):
			UpperCamelCase__				:					str					=					os.path.join(self.test_scripts_folder  ,					'test_performance.py')
			UpperCamelCase__				:					Tuple					=					['accelerate', 'launch', '--num_processes=2', '--num_machines=1', '--machine_rank=0', '--use_fsdp']
			for config in self.performance_configs:
					UpperCamelCase__				:					int					=					cmd.copy()
					for i, strategy in enumerate(UpperCAmelCase_):
							if strategy.lower() in config:
									cmd_config.append(F'--fsdp_sharding_strategy={i+1}')
									break

					if "fp32" in config:
							cmd_config.append('--mixed_precision=no')
					else:
							cmd_config.append('--mixed_precision=fp16')

					if "cpu_offload" in config:
							cmd_config.append('--fsdp_offload_params=True')

					for policy in FSDP_AUTO_WRAP_POLICY:
							if policy.lower() in config:
									cmd_config.append(F'--fsdp_auto_wrap_policy={policy}')
									break

					if policy == "TRANSFORMER_BASED_WRAP":
							cmd_config.append('--fsdp_transformer_layer_cls_to_wrap=BertLayer')
					elif policy == "SIZE_BASED_WRAP":
							cmd_config.append('--fsdp_min_num_params=2000')

					cmd_config.extend(
					    [
					        self.test_file_path,
					        F'--output_dir={self.tmpdir}',
					        F'--performance_lower_bound={self.performance_lower_bound}',
					    ])
					with patch_environment(omp_num_threads=1):
							execute_subprocess_async(UpperCAmelCase_  ,					env=os.environ.copy())

	def      __UpperCamelCase							(      self   :						Any):
			UpperCamelCase__				:					Tuple					=					os.path.join(self.test_scripts_folder  ,					'test_checkpointing.py')
			UpperCamelCase__				:					int					=					[
			    'accelerate',
			    'launch',
			    '--num_processes=2',
			    '--num_machines=1',
			    '--machine_rank=0',
			    '--use_fsdp',
			    '--mixed_precision=fp16',
			    '--fsdp_transformer_layer_cls_to_wrap=BertLayer',
			]

			for i, strategy in enumerate(UpperCAmelCase_):
					UpperCamelCase__				:					Tuple					=					cmd.copy()
					cmd_config.append(F'--fsdp_sharding_strategy={i+1}')
					if strategy != "FULL_SHARD":
							continue
					UpperCamelCase__				:					int					=					len(UpperCAmelCase_)
					for state_dict_type in FSDP_STATE_DICT_TYPE:
							UpperCamelCase__				:					Union[str, Any]					=					cmd_config[:state_dict_config_index]
							cmd_config.append(F'--fsdp_state_dict_type={state_dict_type}')
							cmd_config.extend(
							    [
							        self.test_file_path,
							        F'--output_dir={self.tmpdir}',
							        '--partial_train_epoch=1',
							    ])
							with patch_environment(omp_num_threads=1):
									execute_subprocess_async(UpperCAmelCase_  ,					env=os.environ.copy())

							UpperCamelCase__				:					Optional[Any]					=					cmd_config[:-1]
							UpperCamelCase__				:					Union[str, Any]					=					os.path.join(self.tmpdir  ,					'epoch_0')
							cmd_config.extend(
							    [
							        F'--resume_from_checkpoint={resume_from_checkpoint}',
							    ])
							with patch_environment(omp_num_threads=1):
									execute_subprocess_async(UpperCAmelCase_  ,					env=os.environ.copy())





	def      __UpperCamelCase							(      self   :						Optional[Any]):
			UpperCamelCase__				:					List[str]					=					os.path.join(self.test_scripts_folder  ,					'test_peak_memory_usage.py')
			UpperCamelCase__				:					List[str]					=					[
			    'accelerate',
			    'launch',
			    '--num_processes=2',
			    '--num_machines=1',
			    '--machine_rank=0',
			]
			for spec, peak_mem_upper_bound in self.peak_memory_usage_upper_bound.items():
					UpperCamelCase__				:					int					=					cmd.copy()
					if "fp16" in spec:
							cmd_config.extend(['--mixed_precision=fp16'])
					else:
							cmd_config.extend(['--mixed_precision=no'])

					if "multi_gpu" in spec:
							continue
					else:
							cmd_config.extend(['--use_fsdp'])
							for i, strategy in enumerate(UpperCAmelCase_):
									if strategy.lower() in spec:
											cmd_config.append(F'--fsdp_sharding_strategy={i+1}')
											break

							if "cpu_offload" in spec:
									cmd_config.append('--fsdp_offload_params=True')

							for policy in FSDP_AUTO_WRAP_POLICY:
									if policy.lower() in spec:
											cmd_config.append(F'--fsdp_auto_wrap_policy={policy}')
											break

							if policy == "TRANSFORMER_BASED_WRAP":
									cmd_config.append('--fsdp_transformer_layer_cls_to_wrap=BertLayer')
							elif policy == "SIZE_BASED_WRAP":
									cmd_config.append('--fsdp_min_num_params=2000')

					cmd_config.extend(
					    [
					        self.test_file_path,
					        F'--output_dir={self.tmpdir}',
					        F'--peak_memory_upper_bound={peak_mem_upper_bound}',
					        F'--n_train={self.n_train}',
					        F'--n_val={self.n_val}',
					    ])
					with patch_environment(omp_num_threads=1):
							execute_subprocess_async(UpperCAmelCase_  ,					env=os.environ.copy())












'''simple docstring'''




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   :						Union[str, Any]):
			torch.manual_seed(0)
			UpperCamelCase__				:					Dict					=					TaEncoderModel.from_pretrained('hf-internal-testing/tiny-random-t5')

			torch.manual_seed(0)
			UpperCamelCase__				:					Union[str, Any]					=					AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-t5')

			torch.manual_seed(0)
			UpperCamelCase__				:					List[str]					=					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)
			UpperCamelCase__				:					Optional[Any]					=					DDPMScheduler(
			    num_train_timesteps=1_000  ,					beta_schedule='squaredcos_cap_v2'  ,					beta_start=0.00_01  ,					beta_end=0.02  ,					thresholding=UpperCAmelCase_  ,					dynamic_thresholding_ratio=0.95  ,					sample_max_value=1.0  ,					prediction_type='epsilon'  ,					variance_type='learned_range'  ,					)

			torch.manual_seed(0)
			UpperCamelCase__				:					List[Any]					=					IFWatermarker()

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

	def      __UpperCamelCase							(      self   :						Dict):
			torch.manual_seed(0)
			UpperCamelCase__				:					List[Any]					=					TaEncoderModel.from_pretrained('hf-internal-testing/tiny-random-t5')

			torch.manual_seed(0)
			UpperCamelCase__				:					Any					=					AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-t5')

			torch.manual_seed(0)
			UpperCamelCase__				:					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.4_14  ,					time_embedding_act_fn='gelu'  ,					time_embedding_dim=32  ,					)
			unet.set_attn_processor(AttnAddedKVProcessor())  # For reproducibility tests

			torch.manual_seed(0)
			UpperCamelCase__				:					str					=					DDPMScheduler(
			    num_train_timesteps=1_000  ,					beta_schedule='squaredcos_cap_v2'  ,					beta_start=0.00_01  ,					beta_end=0.02  ,					thresholding=UpperCAmelCase_  ,					dynamic_thresholding_ratio=0.95  ,					sample_max_value=1.0  ,					prediction_type='epsilon'  ,					variance_type='learned_range'  ,					)

			torch.manual_seed(0)
			UpperCamelCase__				:					List[str]					=					DDPMScheduler(
			    num_train_timesteps=1_000  ,					beta_schedule='squaredcos_cap_v2'  ,					beta_start=0.00_01  ,					beta_end=0.02  ,					)

			torch.manual_seed(0)
			UpperCamelCase__				:					Optional[Any]					=					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   :						Any):
			UpperCamelCase__				:					Dict					=					self.get_dummy_components()
			UpperCamelCase__				:					List[Any]					=					self.pipeline_class(**UpperCAmelCase_)
			pipe.to(UpperCAmelCase_)
			pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					Tuple					=					self.get_dummy_inputs(UpperCAmelCase_)

			UpperCamelCase__				:					Optional[Any]					=					inputs['prompt']
			UpperCamelCase__				:					List[Any]					=					inputs['generator']
			UpperCamelCase__				:					Tuple					=					inputs['num_inference_steps']
			UpperCamelCase__				:					List[Any]					=					inputs['output_type']

			if "image" in inputs:
					UpperCamelCase__				:					Tuple					=					inputs['image']
			else:
					UpperCamelCase__				:					Union[str, Any]					=					None

			if "mask_image" in inputs:
					UpperCamelCase__				:					Optional[int]					=					inputs['mask_image']
			else:
					UpperCamelCase__				:					int					=					None

			if "original_image" in inputs:
					UpperCamelCase__				:					List[Any]					=					inputs['original_image']
			else:
					UpperCamelCase__				:					Optional[Any]					=					None

			UpperCamelCase__,	UpperCamelCase__				:					Any					=					pipe.encode_prompt(UpperCAmelCase_)

			# inputs with prompt converted to embeddings
			UpperCamelCase__				:					List[Any]					=					{
			    '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:
					UpperCamelCase__				:					Dict					=					image

			if mask_image is not None:
					UpperCamelCase__				:					Optional[int]					=					mask_image

			if original_image is not None:
					UpperCamelCase__				:					Union[str, Any]					=					original_image

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

			UpperCamelCase__				:					int					=					pipe(**UpperCAmelCase_)[0]

			with tempfile.TemporaryDirectory() as tmpdir:
					pipe.save_pretrained(UpperCAmelCase_)
					UpperCamelCase__				:					Optional[Any]					=					self.pipeline_class.from_pretrained(UpperCAmelCase_)
					pipe_loaded.to(UpperCAmelCase_)
					pipe_loaded.set_progress_bar_config(disable=UpperCAmelCase_)

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

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

			UpperCamelCase__				:					Optional[int]					=					self.get_dummy_inputs(UpperCAmelCase_)

			UpperCamelCase__				:					Union[str, Any]					=					inputs['generator']
			UpperCamelCase__				:					List[Any]					=					inputs['num_inference_steps']
			UpperCamelCase__				:					Optional[int]					=					inputs['output_type']

			# inputs with prompt converted to embeddings
			UpperCamelCase__				:					Any					=					{
			    '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:
					UpperCamelCase__				:					Tuple					=					image

			if mask_image is not None:
					UpperCamelCase__				:					Union[str, Any]					=					mask_image

			if original_image is not None:
					UpperCamelCase__				:					str					=					original_image

			UpperCamelCase__				:					Union[str, Any]					=					pipe_loaded(**UpperCAmelCase_)[0]

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





	def      __UpperCamelCase							(      self   :						Optional[int]):
			UpperCamelCase__				:					Any					=					self.get_dummy_components()
			UpperCamelCase__				:					List[str]					=					self.pipeline_class(**UpperCAmelCase_)
			pipe.to(UpperCAmelCase_)
			pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					Union[str, Any]					=					self.get_dummy_inputs(UpperCAmelCase_)
			UpperCamelCase__				:					Any					=					pipe(**UpperCAmelCase_)[0]

			with tempfile.TemporaryDirectory() as tmpdir:
					pipe.save_pretrained(UpperCAmelCase_)
					UpperCamelCase__				:					Optional[Any]					=					self.pipeline_class.from_pretrained(UpperCAmelCase_)
					pipe_loaded.to(UpperCAmelCase_)
					pipe_loaded.set_progress_bar_config(disable=UpperCAmelCase_)

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

			UpperCamelCase__				:					Any					=					self.get_dummy_inputs(UpperCAmelCase_)
			UpperCamelCase__				:					Tuple					=					pipe_loaded(**UpperCAmelCase_)[0]

			UpperCamelCase__				:					Optional[int]					=					np.abs(to_np(UpperCAmelCase_) - to_np(UpperCAmelCase_)).max()
			self.assertLess(UpperCAmelCase_  ,					1e-4)




