




import tempfile
import unittest

import torch

from diffusers import UNetaDConditionModel
from diffusers.training_utils import EMAModel
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps, torch_device


enable_full_determinism()






class 					__lowerCAmelCase				( unittest.TestCase				):
			snake_case_   :				str				  =  'hf-internal-testing/tiny-stable-diffusion-pipe'
			snake_case_   :				int				  =  1
			snake_case_   :				Any				  =  77
			snake_case_   :				List[Any]				  =  32
			snake_case_   :				List[str]				  =  4
			snake_case_   :				Dict				  =  64
			snake_case_   :				List[str]				  =  torch.manual_seed(0				)







			def UpperCamelCase    ( self					:    Any					,					snake_case__					:    Union[str, Any]=0.9_999   ):




									"""simple docstring"""


									_UpperCAmelCase									=  UNetaDConditionModel.from_pretrained(self.model_id					,					subfolder="unet"   )
									_UpperCAmelCase									=  unet.to(snake_case__   )
									_UpperCAmelCase									=  EMAModel(unet.parameters()					,					decay=snake_case__					,					model_cls=snake_case__					,					model_config=unet.config   )
									return unet, ema_unet







			def UpperCamelCase    ( self					:    Tuple   ):




									"""simple docstring"""


									_UpperCAmelCase									=  torch.randn(
									    self.batch_size					,					self.num_in_channels					,					self.latent_height					,					self.latent_width					,					generator=self.generator   ).to(snake_case__   )
									_UpperCAmelCase									=  torch.randint(0					,					1_000					,					size=(self.batch_size,)					,					generator=self.generator   ).to(snake_case__   )
									_UpperCAmelCase									=  torch.randn(
									    self.batch_size					,					self.prompt_length					,					self.text_encoder_hidden_dim					,					generator=self.generator   ).to(snake_case__   )
									return noisy_latents, timesteps, encoder_hidden_states







			def UpperCamelCase    ( self					:    str					,					snake_case__					:    List[Any]   ):




									"""simple docstring"""


									_UpperCAmelCase									=  {}
									for k, param in unet.state_dict().items():
															_UpperCAmelCase									=  torch.randn_like(snake_case__   ) + (param * torch.randn_like(snake_case__   ))
															updated_state_dict.update({k: updated_param}   )
									unet.load_state_dict(snake_case__   )
									return unet







			def UpperCamelCase    ( self					:    List[Any]   ):




									"""simple docstring"""


									_UpperCAmelCase									=  self.get_models()
									# Take the first (hypothetical) EMA step.
									ema_unet.step(unet.parameters()   )
									assert ema_unet.optimization_step == 1

									# Take two more.
									for _ in range(2   ):
															ema_unet.step(unet.parameters()   )
									assert ema_unet.optimization_step == 3







			def UpperCamelCase    ( self					:    int   ):




									"""simple docstring"""


									_UpperCAmelCase									=  self.get_models()
									# Since the `unet` is not being updated (i.e., backprop'd)
									# there won't be any difference between the `params` of `unet`
									# and `ema_unet` even if we call `ema_unet.step(unet.parameters())`.
									ema_unet.step(unet.parameters()   )
									_UpperCAmelCase									=  list(unet.parameters()   )
									for s_param, param in zip(ema_unet.shadow_params					,					snake_case__   ):
															assert torch.allclose(snake_case__					,					snake_case__   )

									# The above holds true even if we call `ema.step()` multiple times since
									# `unet` params are still not being updated.
									for _ in range(4   ):
															ema_unet.step(unet.parameters()   )
									for s_param, param in zip(ema_unet.shadow_params					,					snake_case__   ):
															assert torch.allclose(snake_case__					,					snake_case__   )







			def UpperCamelCase    ( self					:    Dict   ):




									"""simple docstring"""


									_UpperCAmelCase									=  self.get_models()
									# Here we simulate the parameter updates for `unet`. Since there might
									# be some parameters which are initialized to zero we take extra care to
									# initialize their values to something non-zero before the multiplication.
									_UpperCAmelCase									=  self.simulate_backprop(snake_case__   )

									# Take the EMA step.
									ema_unet.step(unet_pseudo_updated_step_one.parameters()   )

									# Now the EMA'd parameters won't be equal to the original model parameters.
									_UpperCAmelCase									=  list(unet_pseudo_updated_step_one.parameters()   )
									for s_param, param in zip(ema_unet.shadow_params					,					snake_case__   ):
															assert ~torch.allclose(snake_case__					,					snake_case__   )

									# Ensure this is the case when we take multiple EMA steps.
									for _ in range(4   ):
															ema_unet.step(unet.parameters()   )
									for s_param, param in zip(ema_unet.shadow_params					,					snake_case__   ):
															assert ~torch.allclose(snake_case__					,					snake_case__   )







			def UpperCamelCase    ( self					:    Union[str, Any]   ):




									"""simple docstring"""


									_UpperCAmelCase									=  self.get_models()

									# First backprop + EMA
									_UpperCAmelCase									=  self.simulate_backprop(snake_case__   )
									ema_unet.step(unet_step_one.parameters()   )
									_UpperCAmelCase									=  ema_unet.shadow_params

									# Second backprop + EMA
									_UpperCAmelCase									=  self.simulate_backprop(snake_case__   )
									ema_unet.step(unet_step_two.parameters()   )
									_UpperCAmelCase									=  ema_unet.shadow_params

									for step_one, step_two in zip(snake_case__					,					snake_case__   ):
															assert ~torch.allclose(snake_case__					,					snake_case__   )







			def UpperCamelCase    ( self					:    str   ):




									"""simple docstring"""


									_UpperCAmelCase									=  self.get_models(decay=0.0   )
									_UpperCAmelCase									=  self.simulate_backprop(snake_case__   )
									ema_unet.step(unet_step_one.parameters()   )
									_UpperCAmelCase									=  ema_unet.shadow_params

									_UpperCAmelCase									=  self.simulate_backprop(snake_case__   )
									ema_unet.step(unet_step_two.parameters()   )
									_UpperCAmelCase									=  ema_unet.shadow_params

									for step_one, step_two in zip(snake_case__					,					snake_case__   ):
															assert torch.allclose(snake_case__					,					snake_case__   )







			@skip_mps
			def UpperCamelCase    ( self					:    Dict   ):




									"""simple docstring"""


									_UpperCAmelCase									=  self.get_models()
									_UpperCAmelCase									=  self.get_dummy_inputs()

									with tempfile.TemporaryDirectory() as tmpdir:
															ema_unet.save_pretrained(snake_case__   )
															_UpperCAmelCase									=  UNetaDConditionModel.from_pretrained(snake_case__					,					model_cls=snake_case__   )
															_UpperCAmelCase									=  loaded_unet.to(unet.device   )

									# Since no EMA step has been performed the outputs should match.
									_UpperCAmelCase									=  unet(snake_case__					,					snake_case__					,					snake_case__   ).sample
									_UpperCAmelCase									=  loaded_unet(snake_case__					,					snake_case__					,					snake_case__   ).sample

									assert torch.allclose(snake_case__					,					snake_case__					,					atol=1e-4   )









import argparse
import os

import torch
from torchvision.datasets.utils import download_url

from diffusers import AutoencoderKL, DDIMScheduler, DiTPipeline, TransformeraDModel


lowercase_			:       Tuple						       =				{5_12: 'DiT-XL-2-512x512.pt', 2_56: 'DiT-XL-2-256x256.pt'}


def 							__SCREAMING_SNAKE_CASE						(    snake_case_		):

						'''simple docstring'''


						_UpperCAmelCase									=  f"""pretrained_models/{model_name}"""
						if not os.path.isfile(snake_case_		):
												os.makedirs("pretrained_models"			,					exist_ok=snake_case_		)
												_UpperCAmelCase									=  f"""https://dl.fbaipublicfiles.com/DiT/models/{model_name}"""
												download_url(snake_case_			,					"pretrained_models"		)
						_UpperCAmelCase									=  torch.load(snake_case_			,					map_location=lambda snake_case_			,					snake_case_		: storage		)
						return model


def 							__SCREAMING_SNAKE_CASE						(    snake_case_		):

						'''simple docstring'''


						_UpperCAmelCase									=  download_model(pretrained_models[args.image_size]		)

						_UpperCAmelCase									=  state_dict["x_embedder.proj.weight"]
						_UpperCAmelCase									=  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									=  state_dict[
												    "t_embedder.mlp.0.weight"
												]
												_UpperCAmelCase									=  state_dict[
												    "t_embedder.mlp.0.bias"
												]
												_UpperCAmelCase									=  state_dict[
												    "t_embedder.mlp.2.weight"
												]
												_UpperCAmelCase									=  state_dict[
												    "t_embedder.mlp.2.bias"
												]
												_UpperCAmelCase									=  state_dict[
												    "y_embedder.embedding_table.weight"
												]

												_UpperCAmelCase									=  state_dict[
												    f"""blocks.{depth}.adaLN_modulation.1.weight"""
												]
												_UpperCAmelCase									=  state_dict[
												    f"""blocks.{depth}.adaLN_modulation.1.bias"""
												]

												_UpperCAmelCase							,				_UpperCAmelCase							,				_UpperCAmelCase									=  torch.chunk(state_dict[f"""blocks.{depth}.attn.qkv.weight"""]			,					3			,					dim=0		)
												_UpperCAmelCase							,				_UpperCAmelCase							,				_UpperCAmelCase									=  torch.chunk(state_dict[f"""blocks.{depth}.attn.qkv.bias"""]			,					3			,					dim=0		)

												_UpperCAmelCase									=  q
												_UpperCAmelCase									=  q_bias
												_UpperCAmelCase									=  k
												_UpperCAmelCase									=  k_bias
												_UpperCAmelCase									=  v
												_UpperCAmelCase									=  v_bias

												_UpperCAmelCase									=  state_dict[
												    f"""blocks.{depth}.attn.proj.weight"""
												]
												_UpperCAmelCase									=  state_dict[f"""blocks.{depth}.attn.proj.bias"""]

												_UpperCAmelCase									=  state_dict[f"""blocks.{depth}.mlp.fc1.weight"""]
												_UpperCAmelCase									=  state_dict[f"""blocks.{depth}.mlp.fc1.bias"""]
												_UpperCAmelCase									=  state_dict[f"""blocks.{depth}.mlp.fc2.weight"""]
												_UpperCAmelCase									=  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									=  state_dict["final_layer.adaLN_modulation.1.weight"]
						_UpperCAmelCase									=  state_dict["final_layer.adaLN_modulation.1.bias"]
						_UpperCAmelCase									=  state_dict["final_layer.linear.weight"]
						_UpperCAmelCase									=  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									=  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=1000			,					norm_type="ada_norm_zero"			,					norm_elementwise_affine=snake_case_			,					)
						transformer.load_state_dict(snake_case_			,					strict=snake_case_		)

						_UpperCAmelCase									=  DDIMScheduler(
						    num_train_timesteps=1000			,					beta_schedule="linear"			,					prediction_type="epsilon"			,					clip_sample=snake_case_			,					)

						_UpperCAmelCase									=  AutoencoderKL.from_pretrained(args.vae_model		)

						_UpperCAmelCase									=  DiTPipeline(transformer=snake_case_			,					vae=snake_case_			,					scheduler=snake_case_		)

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


if __name__ == "__main__":
				lowercase_			:       Union[str, Any]						       =				argparse.ArgumentParser()

				parser.add_argument(
				    '--image_size',
				    default=2_56,
				    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.'
				)

				lowercase_			:       Tuple						       =				parser.parse_args()
				main(args)




