






'''simple docstring'''







import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import 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


__lowerCAmelCase									=   False





@skip_mps
class 						SCREAMING_SNAKE_CASE       (       SCREAMING_SNAKE_CASE_						, SCREAMING_SNAKE_CASE_						, SCREAMING_SNAKE_CASE_						, unittest.TestCase							):
			snake_case__	=					StableDiffusionAttendAndExcitePipeline
			snake_case__	=					False
			snake_case__	=					TEXT_TO_IMAGE_PARAMS
			snake_case__	=					TEXT_TO_IMAGE_BATCH_PARAMS.union({"token_indices"}							)
			snake_case__	=					TEXT_TO_IMAGE_IMAGE_PARAMS
			snake_case__	=					TEXT_TO_IMAGE_IMAGE_PARAMS






			@classmethod
			def       SCREAMING_SNAKE_CASE   (			cls   :      Tuple       )      ->      List[Any]:
							super().setUpClass()
							torch.use_deterministic_algorithms(__SCREAMING_SNAKE_CASE       )






			@classmethod
			def       SCREAMING_SNAKE_CASE   (			cls   :      Optional[int]       )      ->      List[str]:
							super().tearDownClass()
							torch.use_deterministic_algorithms(__SCREAMING_SNAKE_CASE       )






			def       SCREAMING_SNAKE_CASE   (			self   :      int       )      ->      Union[str, Any]:
							torch.manual_seed(0       )
							a_		:	str									=   UNetaDConditionModel(
							    block_out_channels=(32, 64)	,    layers_per_block=1	,    sample_size=32	,    in_channels=4	,    out_channels=4	,    down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')	,    up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')	,    cross_attention_dim=32	,    attention_head_dim=(2, 4)	,    use_linear_projection=__SCREAMING_SNAKE_CASE	,    )
							a_		:	Optional[Any]									=   DDIMScheduler(
							    beta_start=0.0_0085	,    beta_end=0.012	,    beta_schedule='''scaled_linear'''	,    clip_sample=__SCREAMING_SNAKE_CASE	,    set_alpha_to_one=__SCREAMING_SNAKE_CASE	,    )
							torch.manual_seed(0       )
							a_		:	Optional[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	,    sample_size=128	,    )
							torch.manual_seed(0       )
							a_		:	Optional[int]									=   CLIPTextConfig(
							    bos_token_id=0	,    eos_token_id=2	,    hidden_size=32	,    intermediate_size=37	,    layer_norm_eps=1e-05	,    num_attention_heads=4	,    num_hidden_layers=5	,    pad_token_id=1	,    vocab_size=1000	,    hidden_act='''gelu'''	,    projection_dim=512	,    )
							a_		:	Tuple									=   CLIPTextModel(__SCREAMING_SNAKE_CASE       )
							a_		:	Any									=   CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''       )

							a_		:	Optional[int]									=   {
							    '''unet''': unet,
							    '''scheduler''': scheduler,
							    '''vae''': vae,
							    '''text_encoder''': text_encoder,
							    '''tokenizer''': tokenizer,
							    '''safety_checker''': None,
							    '''feature_extractor''': None,
							}

							return components






			def       SCREAMING_SNAKE_CASE   (			self   :      List[str]	,    __SCREAMING_SNAKE_CASE   :      Optional[Any]	,    __SCREAMING_SNAKE_CASE   :      int=0       )      ->      Optional[Any]:
							if str(__SCREAMING_SNAKE_CASE       ).startswith('''mps'''       ):
											a_		:	Any									=   torch.manual_seed(__SCREAMING_SNAKE_CASE       )
							else:
											a_		:	List[str]									=   torch.Generator(device=__SCREAMING_SNAKE_CASE       ).manual_seed(__SCREAMING_SNAKE_CASE       )
							a_		:	str									=   {
							    '''prompt''': '''a cat and a frog''',
							    '''token_indices''': [2, 5],
							    '''generator''': generator,
							    '''num_inference_steps''': 1,
							    '''guidance_scale''': 6.0,
							    '''output_type''': '''numpy''',
							    '''max_iter_to_alter''': 2,
							    '''thresholds''': {0: 0.7},
							}
							return inputs






			def       SCREAMING_SNAKE_CASE   (			self   :      Union[str, Any]       )      ->      Tuple:
							a_		:	List[str]									=   '''cpu'''

							a_		:	Union[str, Any]									=   self.get_dummy_components()
							a_		:	Optional[int]									=   self.pipeline_class(**__SCREAMING_SNAKE_CASE       )
							pipe.to(__SCREAMING_SNAKE_CASE       )
							pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

							a_		:	List[str]									=   self.get_dummy_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	Optional[int]									=   pipe(**__SCREAMING_SNAKE_CASE       ).images
							a_		:	Optional[int]									=   image[0, -3:, -3:, -1]

							self.assertEqual(image.shape	,    (1, 64, 64, 3)       )
							a_		:	Union[str, Any]									=   np.array(
							    [0.6390_5364, 0.6289_7307, 0.4859_9017, 0.513_3624, 0.555_0048, 0.4576_9516, 0.5032_6973, 0.502_3139, 0.4538_4496]       )
							a_		:	List[str]									=   np.abs(image_slice.flatten() - expected_slice       ).max()
							self.assertLessEqual(__SCREAMING_SNAKE_CASE	,    1e-3       )






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[Any]       )      ->      List[str]:
							super().test_cpu_offload_forward_pass(expected_max_diff=5e-4       )






			def       SCREAMING_SNAKE_CASE   (			self   :      Dict       )      ->      Optional[int]:
							# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
							self._test_inference_batch_consistent(batch_sizes=[1, 2]       )






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[int]       )      ->      str:
							self._test_inference_batch_single_identical(batch_size=2	,    expected_max_diff=7e-4       )






			def       SCREAMING_SNAKE_CASE   (			self   :      Tuple       )      ->      Any:
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3       )






			def       SCREAMING_SNAKE_CASE   (			self   :      List[Any]       )      ->      Optional[Any]:
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4       )






			def       SCREAMING_SNAKE_CASE   (			self   :      Union[str, Any]       )      ->      Tuple:
							super().test_save_load_local(expected_max_difference=5e-4       )







			def       SCREAMING_SNAKE_CASE   (			self   :      Tuple       )      ->      List[Any]:
							super().test_save_load_optional_components(expected_max_difference=4e-4       )





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






			@classmethod
			def       SCREAMING_SNAKE_CASE   (			cls   :      Dict       )      ->      Optional[int]:
							super().setUpClass()
							torch.use_deterministic_algorithms(__SCREAMING_SNAKE_CASE       )






			@classmethod
			def       SCREAMING_SNAKE_CASE   (			cls   :      Any       )      ->      Union[str, Any]:
							super().tearDownClass()
							torch.use_deterministic_algorithms(__SCREAMING_SNAKE_CASE       )






			def       SCREAMING_SNAKE_CASE   (			self   :      Tuple       )      ->      List[str]:
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()







			def       SCREAMING_SNAKE_CASE   (			self   :      str       )      ->      Tuple:
							a_		:	List[Any]									=   torch.manual_seed(51       )

							a_		:	str									=   StableDiffusionAttendAndExcitePipeline.from_pretrained(
							    '''CompVis/stable-diffusion-v1-4'''	,    safety_checker=__SCREAMING_SNAKE_CASE	,    torch_dtype=torch.floataa       )
							pipe.to('''cuda'''       )

							a_		:	Tuple									=   '''a painting of an elephant with glasses'''
							a_		:	Optional[int]									=   [5, 7]

							a_		:	Dict									=   pipe(
							    prompt=__SCREAMING_SNAKE_CASE	,    token_indices=__SCREAMING_SNAKE_CASE	,    guidance_scale=7.5	,    generator=__SCREAMING_SNAKE_CASE	,    num_inference_steps=5	,    max_iter_to_alter=5	,    output_type='''numpy'''	,    ).images[0]

							a_		:	Optional[int]									=   load_numpy(
							    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy'''       )
							assert np.abs((expected_image - image).max()       ) < 5e-1












'''simple docstring'''







import argparse
import os

import torch

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)


__lowerCAmelCase									=   {
    'sample_size': 32,
    'in_channels': 3,
    'out_channels': 3,
    'layers_per_block': 2,
    'num_class_embeds': 1_000,
    'block_out_channels': [32, 64],
    'attention_head_dim': 8,
    'down_block_types': [
        'ResnetDownsampleBlock2D',
        'AttnDownBlock2D',
    ],
    'up_block_types': [
        'AttnUpBlock2D',
        'ResnetUpsampleBlock2D',
    ],
    'resnet_time_scale_shift': 'scale_shift',
    'upsample_type': 'resnet',
    'downsample_type': 'resnet',
}

__lowerCAmelCase									=   {
    'sample_size': 64,
    'in_channels': 3,
    'out_channels': 3,
    'layers_per_block': 3,
    'num_class_embeds': 1_000,
    'block_out_channels': [192, 192 * 2, 192 * 3, 192 * 4],
    'attention_head_dim': 64,
    'down_block_types': [
        'ResnetDownsampleBlock2D',
        'AttnDownBlock2D',
        'AttnDownBlock2D',
        'AttnDownBlock2D',
    ],
    'up_block_types': [
        'AttnUpBlock2D',
        'AttnUpBlock2D',
        'AttnUpBlock2D',
        'ResnetUpsampleBlock2D',
    ],
    'resnet_time_scale_shift': 'scale_shift',
    'upsample_type': 'resnet',
    'downsample_type': 'resnet',
}

__lowerCAmelCase									=   {
    'sample_size': 256,
    'in_channels': 3,
    'out_channels': 3,
    'layers_per_block': 2,
    'num_class_embeds': None,
    'block_out_channels': [256, 256, 256 * 2, 256 * 2, 256 * 4, 256 * 4],
    'attention_head_dim': 64,
    'down_block_types': [
        'ResnetDownsampleBlock2D',
        'ResnetDownsampleBlock2D',
        'ResnetDownsampleBlock2D',
        'AttnDownBlock2D',
        'AttnDownBlock2D',
        'AttnDownBlock2D',
    ],
    'up_block_types': [
        'AttnUpBlock2D',
        'AttnUpBlock2D',
        'AttnUpBlock2D',
        'ResnetUpsampleBlock2D',
        'ResnetUpsampleBlock2D',
        'ResnetUpsampleBlock2D',
    ],
    'resnet_time_scale_shift': 'default',
    'upsample_type': 'resnet',
    'downsample_type': 'resnet',
}

__lowerCAmelCase									=   {
    'num_train_timesteps': 40,
    'sigma_min': 0.002,
    'sigma_max': 80.0,
}

__lowerCAmelCase									=   {
    'num_train_timesteps': 201,
    'sigma_min': 0.002,
    'sigma_max': 80.0,
}

__lowerCAmelCase									=   {
    'num_train_timesteps': 151,
    'sigma_min': 0.002,
    'sigma_max': 80.0,
}





def 		_UpperCAmelCase (				__A				: Tuple   ):
				if isinstance(__A    ,		__A   ):
								return v
				if v.lower() in ("yes", "true", "t", "y", "1"):
								return True
				elif v.lower() in ("no", "false", "f", "n", "0"):
								return False
				else:
								raise argparse.ArgumentTypeError('''boolean value expected'''   )





def 		_UpperCAmelCase (				__A				: str    ,		__A				: Tuple    ,		__A				: Optional[int]    ,		__A				: int    ,		__A				: Optional[Any]=False   ):
				a_		:	str									=   checkpoint[f'{old_prefix}.in_layers.0.weight']
				a_		:	Dict									=   checkpoint[f'{old_prefix}.in_layers.0.bias']
				a_		:	Optional[Any]									=   checkpoint[f'{old_prefix}.in_layers.2.weight']
				a_		:	Any									=   checkpoint[f'{old_prefix}.in_layers.2.bias']
				a_		:	Any									=   checkpoint[f'{old_prefix}.emb_layers.1.weight']
				a_		:	int									=   checkpoint[f'{old_prefix}.emb_layers.1.bias']
				a_		:	Union[str, Any]									=   checkpoint[f'{old_prefix}.out_layers.0.weight']
				a_		:	Optional[Any]									=   checkpoint[f'{old_prefix}.out_layers.0.bias']
				a_		:	Optional[Any]									=   checkpoint[f'{old_prefix}.out_layers.3.weight']
				a_		:	List[Any]									=   checkpoint[f'{old_prefix}.out_layers.3.bias']

				if has_skip:
								a_		:	int									=   checkpoint[f'{old_prefix}.skip_connection.weight']
								a_		:	List[Any]									=   checkpoint[f'{old_prefix}.skip_connection.bias']

				return new_checkpoint





def 		_UpperCAmelCase (				__A				: int    ,		__A				: int    ,		__A				: Union[str, Any]    ,		__A				: Optional[int]    ,		__A				: List[str]=None   ):
				a_					,      a_					,      a_		:	Any									=   checkpoint[f'{old_prefix}.qkv.weight'].chunk(3    ,		dim=0   )
				a_					,      a_					,      a_		:	List[Any]									=   checkpoint[f'{old_prefix}.qkv.bias'].chunk(3    ,		dim=0   )

				a_		:	List[str]									=   checkpoint[f'{old_prefix}.norm.weight']
				a_		:	Optional[Any]									=   checkpoint[f'{old_prefix}.norm.bias']

				a_		:	Any									=   weight_q.squeeze(-1   ).squeeze(-1   )
				a_		:	int									=   bias_q.squeeze(-1   ).squeeze(-1   )
				a_		:	Any									=   weight_k.squeeze(-1   ).squeeze(-1   )
				a_		:	Optional[int]									=   bias_k.squeeze(-1   ).squeeze(-1   )
				a_		:	str									=   weight_v.squeeze(-1   ).squeeze(-1   )
				a_		:	Optional[Any]									=   bias_v.squeeze(-1   ).squeeze(-1   )

				a_		:	Union[str, Any]									=   (
				    checkpoint[f'{old_prefix}.proj_out.weight'].squeeze(-1   ).squeeze(-1   )
				)
				a_		:	Any									=   checkpoint[f'{old_prefix}.proj_out.bias'].squeeze(-1   ).squeeze(-1   )

				return new_checkpoint





def 		_UpperCAmelCase (				__A				: str    ,		__A				: str   ):
				a_		:	int									=   torch.load(__A    ,		map_location='''cpu'''   )
				a_		:	Union[str, Any]									=   {}

				a_		:	int									=   checkpoint['''time_embed.0.weight''']
				a_		:	Tuple									=   checkpoint['''time_embed.0.bias''']
				a_		:	List[Any]									=   checkpoint['''time_embed.2.weight''']
				a_		:	Union[str, Any]									=   checkpoint['''time_embed.2.bias''']

				if unet_config["num_class_embeds"] is not None:
								a_		:	List[Any]									=   checkpoint['''label_emb.weight''']

				a_		:	Optional[int]									=   checkpoint['''input_blocks.0.0.weight''']
				a_		:	int									=   checkpoint['''input_blocks.0.0.bias''']

				a_		:	Optional[int]									=   unet_config['''down_block_types''']
				a_		:	Union[str, Any]									=   unet_config['''layers_per_block''']
				a_		:	Optional[Any]									=   unet_config['''attention_head_dim''']
				a_		:	str									=   unet_config['''block_out_channels''']
				a_		:	Any									=   1
				a_		:	Union[str, Any]									=   channels_list[0]

				for i, layer_type in enumerate(__A   ):
								a_		:	Any									=   channels_list[i]
								a_		:	Tuple									=   current_channels != prev_channels
								if layer_type == "ResnetDownsampleBlock2D":
												for j in range(__A   ):
																a_		:	Union[str, Any]									=   f'down_blocks.{i}.resnets.{j}'
																a_		:	int									=   f'input_blocks.{current_layer}.0'
																a_		:	List[Any]									=   True if j == 0 and downsample_block_has_skip else False
																a_		:	List[str]									=   convert_resnet(__A    ,		__A    ,		__A    ,		__A    ,		has_skip=__A   )
																current_layer += 1

								elif layer_type == "AttnDownBlock2D":
												for j in range(__A   ):
																a_		:	Union[str, Any]									=   f'down_blocks.{i}.resnets.{j}'
																a_		:	Optional[int]									=   f'input_blocks.{current_layer}.0'
																a_		:	Optional[int]									=   True if j == 0 and downsample_block_has_skip else False
																a_		:	Any									=   convert_resnet(__A    ,		__A    ,		__A    ,		__A    ,		has_skip=__A   )
																a_		:	Optional[int]									=   f'down_blocks.{i}.attentions.{j}'
																a_		:	List[Any]									=   f'input_blocks.{current_layer}.1'
																a_		:	int									=   convert_attention(
																    __A    ,		__A    ,		__A    ,		__A    ,		__A   )
																current_layer += 1

								if i != len(__A   ) - 1:
												a_		:	Dict									=   f'down_blocks.{i}.downsamplers.0'
												a_		:	Optional[int]									=   f'input_blocks.{current_layer}.0'
												a_		:	Union[str, Any]									=   convert_resnet(__A    ,		__A    ,		__A    ,		__A   )
												current_layer += 1

								a_		:	Any									=   current_channels

				# hardcoded the mid-block for now
				a_		:	Tuple									=   '''mid_block.resnets.0'''
				a_		:	int									=   '''middle_block.0'''
				a_		:	Tuple									=   convert_resnet(__A    ,		__A    ,		__A    ,		__A   )
				a_		:	List[Any]									=   '''mid_block.attentions.0'''
				a_		:	Optional[int]									=   '''middle_block.1'''
				a_		:	Optional[int]									=   convert_attention(__A    ,		__A    ,		__A    ,		__A    ,		__A   )
				a_		:	Any									=   '''mid_block.resnets.1'''
				a_		:	Tuple									=   '''middle_block.2'''
				a_		:	Tuple									=   convert_resnet(__A    ,		__A    ,		__A    ,		__A   )

				a_		:	str									=   0
				a_		:	Optional[int]									=   unet_config['''up_block_types''']

				for i, layer_type in enumerate(__A   ):
								if layer_type == "ResnetUpsampleBlock2D":
												for j in range(layers_per_block + 1   ):
																a_		:	Tuple									=   f'up_blocks.{i}.resnets.{j}'
																a_		:	Optional[int]									=   f'output_blocks.{current_layer}.0'
																a_		:	Optional[Any]									=   convert_resnet(__A    ,		__A    ,		__A    ,		__A    ,		has_skip=__A   )
																current_layer += 1

												if i != len(__A   ) - 1:
																a_		:	int									=   f'up_blocks.{i}.upsamplers.0'
																a_		:	Union[str, Any]									=   f'output_blocks.{current_layer-1}.1'
																a_		:	int									=   convert_resnet(__A    ,		__A    ,		__A    ,		__A   )
								elif layer_type == "AttnUpBlock2D":
												for j in range(layers_per_block + 1   ):
																a_		:	Optional[Any]									=   f'up_blocks.{i}.resnets.{j}'
																a_		:	str									=   f'output_blocks.{current_layer}.0'
																a_		:	List[Any]									=   convert_resnet(__A    ,		__A    ,		__A    ,		__A    ,		has_skip=__A   )
																a_		:	List[Any]									=   f'up_blocks.{i}.attentions.{j}'
																a_		:	List[str]									=   f'output_blocks.{current_layer}.1'
																a_		:	str									=   convert_attention(
																    __A    ,		__A    ,		__A    ,		__A    ,		__A   )
																current_layer += 1

												if i != len(__A   ) - 1:
																a_		:	Dict									=   f'up_blocks.{i}.upsamplers.0'
																a_		:	Any									=   f'output_blocks.{current_layer-1}.2'
																a_		:	int									=   convert_resnet(__A    ,		__A    ,		__A    ,		__A   )

				a_		:	Dict									=   checkpoint['''out.0.weight''']
				a_		:	int									=   checkpoint['''out.0.bias''']
				a_		:	int									=   checkpoint['''out.2.weight''']
				a_		:	str									=   checkpoint['''out.2.bias''']

				return new_checkpoint


if __name__ == "__main__":
		__lowerCAmelCase									=   argparse.ArgumentParser()

		parser.add_argument('--unet_path', default=None, type=str, required=True, help='Path to the unet.pt to convert.')
		parser.add_argument(
		    '--dump_path', default=None, type=str, required=True, help='Path to output the converted UNet model.'
		)
		parser.add_argument('--class_cond', default=True, type=str, help='Whether the model is class-conditional.')

		__lowerCAmelCase									=   parser.parse_args()
		__lowerCAmelCase									=   strabool(args.class_cond)

		__lowerCAmelCase									=   os.path.basename(args.unet_path)
		print(F"""Checkpoint: {ckpt_name}""")

		# Get U-Net config
		if "imagenet64" in ckpt_name:
				__lowerCAmelCase									=   IMAGENET_64_UNET_CONFIG
		elif "256" in ckpt_name and (("bedroom" in ckpt_name) or ("cat" in ckpt_name)):
				__lowerCAmelCase									=   LSUN_256_UNET_CONFIG
		elif "test" in ckpt_name:
				__lowerCAmelCase									=   TEST_UNET_CONFIG
		else:
				raise ValueError(F"""Checkpoint type {ckpt_name} is not currently supported.""")

		if not args.class_cond:
				__lowerCAmelCase									=   None

		__lowerCAmelCase									=   con_pt_to_diffuser(args.unet_path, unet_config)

		__lowerCAmelCase									=   UNetaDModel(**unet_config)
		image_unet.load_state_dict(converted_unet_ckpt)

		# Get scheduler config
		if "cd" in ckpt_name or "test" in ckpt_name:
				__lowerCAmelCase									=   CD_SCHEDULER_CONFIG
		elif "ct" in ckpt_name and "imagenet64" in ckpt_name:
				__lowerCAmelCase									=   CT_IMAGENET_64_SCHEDULER_CONFIG
		elif "ct" in ckpt_name and "256" in ckpt_name and (("bedroom" in ckpt_name) or ("cat" in ckpt_name)):
				__lowerCAmelCase									=   CT_LSUN_256_SCHEDULER_CONFIG
		else:
				raise ValueError(F"""Checkpoint type {ckpt_name} is not currently supported.""")

		__lowerCAmelCase									=   CMStochasticIterativeScheduler(**scheduler_config)

		__lowerCAmelCase									=   ConsistencyModelPipeline(unet=image_unet, scheduler=cm_scheduler)
		consistency_model.save_pretrained(args.dump_path)




