




import tempfile
import unittest

import numpy as np
import torch

from diffusers import DiffusionPipeline
from diffusers.models.attention_processor import Attention, AttnAddedKVProcessor


class       SCREAMING_SNAKE_CASE_							(					unittest.TestCase      ):



	"""simple docstring"""


	def 				lowerCamelCase__    (				self		:	Any				,    lowerCAmelCase		:	bool = False  )  ->					Any:







						"""simple docstring"""







						__UpperCamelCase				:  Dict								=			10

						if only_cross_attention:
											__UpperCamelCase				:  Tuple								=			12
						else:
											# when only cross attention is not set, the cross attention dim must be the same as the query dim
											__UpperCamelCase				:  Optional[int]								=			query_dim

						return {
						    "query_dim": query_dim,
						    "cross_attention_dim": cross_attention_dim,
						    "heads": 2,
						    "dim_head": 4,
						    "added_kv_proj_dim": 6,
						    "norm_num_groups": 1,
						    "only_cross_attention": only_cross_attention,
						    "processor": AttnAddedKVProcessor(),
						}






	def 				lowerCamelCase__    (				self		:	Union[str, Any]				,    lowerCAmelCase		:	Tuple				,    lowerCAmelCase		:	Any  )  ->					Dict:







						"""simple docstring"""







						__UpperCamelCase				:  Tuple								=			2

						__UpperCamelCase				:  str								=			torch.rand(lowerCAmelCase				,    lowerCAmelCase				,    3				,    2  )
						__UpperCamelCase				:  Optional[int]								=			torch.rand(lowerCAmelCase				,    4				,    lowerCAmelCase  )
						__UpperCamelCase				:  int								=			None

						return {
						    "hidden_states": hidden_states,
						    "encoder_hidden_states": encoder_hidden_states,
						    "attention_mask": attention_mask,
						}






	def 				lowerCamelCase__    (				self		:	Optional[Any]  )  ->					Optional[Any]:







						"""simple docstring"""







						torch.manual_seed(0  )

						__UpperCamelCase				:  List[str]								=			self.get_constructor_arguments(only_cross_attention=lowerCAmelCase  )
						__UpperCamelCase				:  List[str]								=			Attention(**lowerCAmelCase  )

						self.assertTrue(attn.to_k is not None  )
						self.assertTrue(attn.to_v is not None  )

						__UpperCamelCase				:  Dict								=			self.get_forward_arguments(
						    query_dim=constructor_args["""query_dim"""]				,    added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]  )

						__UpperCamelCase				:  Optional[Any]								=			attn(**lowerCAmelCase  )

						# only self attention

						torch.manual_seed(0  )

						__UpperCamelCase				:  Any								=			self.get_constructor_arguments(only_cross_attention=lowerCAmelCase  )
						__UpperCamelCase				:  List[str]								=			Attention(**lowerCAmelCase  )

						self.assertTrue(attn.to_k is None  )
						self.assertTrue(attn.to_v is None  )

						__UpperCamelCase				:  List[Any]								=			self.get_forward_arguments(
						    query_dim=constructor_args["""query_dim"""]				,    added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]  )

						__UpperCamelCase				:  List[str]								=			attn(**lowerCAmelCase  )

						self.assertTrue((only_cross_attn_out != self_and_cross_attn_out).all()  )






class       SCREAMING_SNAKE_CASE_							(					unittest.TestCase      ):



	"""simple docstring"""


	def 				lowerCamelCase__    (				self		:	Optional[int]  )  ->					Tuple:







						"""simple docstring"""







						__UpperCamelCase				:  Optional[int]								=			DiffusionPipeline.from_pretrained("""hf-internal-testing/tiny-stable-diffusion-pipe"""				,    safety_checker=lowerCAmelCase  )

						__UpperCamelCase				:  List[Any]								=			pipe(
						    """foo"""				,    num_inference_steps=2				,    generator=torch.Generator("""cpu"""  ).manual_seed(0  )				,    output_type="""np"""				,    ).images

						# the initial conversion succeeds
						__UpperCamelCase				:  Union[str, Any]								=			DiffusionPipeline.from_pretrained(
						    """hf-internal-testing/tiny-stable-diffusion-pipe"""				,    device_map="""sequential"""				,    safety_checker=lowerCAmelCase  )

						__UpperCamelCase				:  Dict								=			pipe(
						    """foo"""				,    num_inference_steps=2				,    generator=torch.Generator("""cpu"""  ).manual_seed(0  )				,    output_type="""np"""				,    ).images

						with tempfile.TemporaryDirectory() as tmpdir:
											# save the converted model
											pipe.save_pretrained(lowerCAmelCase  )

											# can also load the converted weights
											__UpperCamelCase				:  Optional[int]								=			DiffusionPipeline.from_pretrained(lowerCAmelCase				,    device_map="""sequential"""				,    safety_checker=lowerCAmelCase  )

						__UpperCamelCase				:  int								=			pipe(
						    """foo"""				,    num_inference_steps=2				,    generator=torch.Generator("""cpu"""  ).manual_seed(0  )				,    output_type="""np"""				,    ).images

						self.assertTrue(np.allclose(lowerCAmelCase				,    lowerCAmelCase				,    atol=1E-5  )  )
						self.assertTrue(np.allclose(lowerCAmelCase				,    lowerCAmelCase				,    atol=1E-5  )  )












import argparse
import os

import torch
from torchvision.datasets.utils import download_url

from diffusers import AutoencoderKL, DDIMScheduler, DiTPipeline, TransformeraDModel


a__		    =    {512: '''DiT-XL-2-512x512.pt''', 256: '''DiT-XL-2-256x256.pt'''}





def 							A__   (snake_case      :		List[Any]							)    ->	Dict:
					__UpperCamelCase				:  Optional[int]								=			F'''pretrained_models/{model_name}'''
					if not os.path.isfile(snake_case							):
										os.makedirs("""pretrained_models"""		,  exist_ok=snake_case							)
										__UpperCamelCase				:  Optional[int]								=			F'''https://dl.fbaipublicfiles.com/DiT/models/{model_name}'''
										download_url(snake_case		,  """pretrained_models"""							)
					__UpperCamelCase				:  Union[str, Any]								=			torch.load(snake_case		,  map_location=lambda snake_case		,  snake_case							: storage							)
					return model





def 							A__   (snake_case      :		Optional[Any]							)    ->	Union[str, Any]:
					__UpperCamelCase				:  Any								=			download_model(pretrained_models[args.image_size]							)

					__UpperCamelCase				:  Optional[Any]								=			state_dict["""x_embedder.proj.weight"""]
					__UpperCamelCase				:  List[str]								=			state_dict["""x_embedder.proj.bias"""]
					state_dict.pop("""x_embedder.proj.weight"""							)
					state_dict.pop("""x_embedder.proj.bias"""							)

					for depth in range(28							):
										__UpperCamelCase				:  Tuple								=			state_dict[
										    """t_embedder.mlp.0.weight"""
										]
										__UpperCamelCase				:  Optional[Any]								=			state_dict[
										    """t_embedder.mlp.0.bias"""
										]
										__UpperCamelCase				:  Any								=			state_dict[
										    """t_embedder.mlp.2.weight"""
										]
										__UpperCamelCase				:  Optional[Any]								=			state_dict[
										    """t_embedder.mlp.2.bias"""
										]
										__UpperCamelCase				:  Optional[int]								=			state_dict[
										    """y_embedder.embedding_table.weight"""
										]

										__UpperCamelCase				:  List[str]								=			state_dict[
										    F'''blocks.{depth}.adaLN_modulation.1.weight'''
										]
										__UpperCamelCase				:  Union[str, Any]								=			state_dict[
										    F'''blocks.{depth}.adaLN_modulation.1.bias'''
										]

										__UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase				:  str								=			torch.chunk(state_dict[F'''blocks.{depth}.attn.qkv.weight''']		,  3		,  dim=0							)
										__UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase				:  Any								=			torch.chunk(state_dict[F'''blocks.{depth}.attn.qkv.bias''']		,  3		,  dim=0							)

										__UpperCamelCase				:  Optional[Any]								=			q
										__UpperCamelCase				:  Tuple								=			q_bias
										__UpperCamelCase				:  List[str]								=			k
										__UpperCamelCase				:  Tuple								=			k_bias
										__UpperCamelCase				:  List[Any]								=			v
										__UpperCamelCase				:  Dict								=			v_bias

										__UpperCamelCase				:  Tuple								=			state_dict[
										    F'''blocks.{depth}.attn.proj.weight'''
										]
										__UpperCamelCase				:  Union[str, Any]								=			state_dict[F'''blocks.{depth}.attn.proj.bias''']

										__UpperCamelCase				:  List[Any]								=			state_dict[F'''blocks.{depth}.mlp.fc1.weight''']
										__UpperCamelCase				:  List[str]								=			state_dict[F'''blocks.{depth}.mlp.fc1.bias''']
										__UpperCamelCase				:  Optional[Any]								=			state_dict[F'''blocks.{depth}.mlp.fc2.weight''']
										__UpperCamelCase				:  int								=			state_dict[F'''blocks.{depth}.mlp.fc2.bias''']

										state_dict.pop(F'''blocks.{depth}.attn.qkv.weight'''							)
										state_dict.pop(F'''blocks.{depth}.attn.qkv.bias'''							)
										state_dict.pop(F'''blocks.{depth}.attn.proj.weight'''							)
										state_dict.pop(F'''blocks.{depth}.attn.proj.bias'''							)
										state_dict.pop(F'''blocks.{depth}.mlp.fc1.weight'''							)
										state_dict.pop(F'''blocks.{depth}.mlp.fc1.bias'''							)
										state_dict.pop(F'''blocks.{depth}.mlp.fc2.weight'''							)
										state_dict.pop(F'''blocks.{depth}.mlp.fc2.bias'''							)
										state_dict.pop(F'''blocks.{depth}.adaLN_modulation.1.weight'''							)
										state_dict.pop(F'''blocks.{depth}.adaLN_modulation.1.bias'''							)

					state_dict.pop("""t_embedder.mlp.0.weight"""							)
					state_dict.pop("""t_embedder.mlp.0.bias"""							)
					state_dict.pop("""t_embedder.mlp.2.weight"""							)
					state_dict.pop("""t_embedder.mlp.2.bias"""							)
					state_dict.pop("""y_embedder.embedding_table.weight"""							)

					__UpperCamelCase				:  str								=			state_dict["""final_layer.adaLN_modulation.1.weight"""]
					__UpperCamelCase				:  List[str]								=			state_dict["""final_layer.adaLN_modulation.1.bias"""]
					__UpperCamelCase				:  Union[str, Any]								=			state_dict["""final_layer.linear.weight"""]
					__UpperCamelCase				:  Any								=			state_dict["""final_layer.linear.bias"""]

					state_dict.pop("""final_layer.linear.weight"""							)
					state_dict.pop("""final_layer.linear.bias"""							)
					state_dict.pop("""final_layer.adaLN_modulation.1.weight"""							)
					state_dict.pop("""final_layer.adaLN_modulation.1.bias"""							)

					# DiT XL/2
					__UpperCamelCase				:  Optional[Any]								=			TransformeraDModel(
					    sample_size=args.image_size // 8		,  num_layers=28		,  attention_head_dim=72		,  in_channels=4		,  out_channels=8		,  patch_size=2		,  attention_bias=snake_case		,  num_attention_heads=16		,  activation_fn="""gelu-approximate"""		,  num_embeds_ada_norm=10_00		,  norm_type="""ada_norm_zero"""		,  norm_elementwise_affine=snake_case		,  )
					transformer.load_state_dict(snake_case		,  strict=snake_case							)

					__UpperCamelCase				:  Union[str, Any]								=			DDIMScheduler(
					    num_train_timesteps=10_00		,  beta_schedule="""linear"""		,  prediction_type="""epsilon"""		,  clip_sample=snake_case		,  )

					__UpperCamelCase				:  Optional[int]								=			AutoencoderKL.from_pretrained(args.vae_model							)

					__UpperCamelCase				:  str								=			DiTPipeline(transformer=snake_case		,  vae=snake_case		,  scheduler=snake_case							)

					if args.save:
										pipeline.save_pretrained(args.checkpoint_path							)


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

	parser.add_argument(
	    '''--image_size''',
	    default=256,
	    type=int,
	    required=False,
	    help='''Image size of pretrained model, either 256 or 512.''',
	)
	parser.add_argument(
	    '''--vae_model''',
	    default='''stabilityai/sd-vae-ft-ema''',
	    type=str,
	    required=False,
	    help='''Path to pretrained VAE model, either stabilityai/sd-vae-ft-mse or stabilityai/sd-vae-ft-ema.''',
	)
	parser.add_argument(
	    '''--save''', default=True, type=bool, required=False, help='''Whether to save the converted pipeline or not.'''
	)
	parser.add_argument(
	    '''--checkpoint_path''', default=None, type=str, required=True, help='''Path to the output pipeline.'''
	)

	a__		    =    parser.parse_args()
	main(args)






