




import json
import os
from collections.abc import Mapping
from typing import Dict, List, Optional, Union

import numpy as np
import torch

from .imports import is_safetensors_available





def 							A__   (snake_case      :		Tuple		,  snake_case      :		str		,  snake_case      :		Optional[Any]		,  snake_case      :		Optional[Any]=None							)    ->	Union[str, Any]:
					__UpperCamelCase				:  Union[str, Any]								=			None
					# Check the string instead of the dtype to be compatible with versions of PyTorch that don't have bfloat16.
					if str(weight.dtype							) == "torch.bfloat16":
										# Need to reinterpret the underlined data as int16 since NumPy does not handle bfloat16s.
										__UpperCamelCase				:  Optional[int]								=			weight.view(torch.intaa							)
										__UpperCamelCase				:  str								=			"""bfloat16"""
					__UpperCamelCase				:  Tuple								=			weight.cpu().numpy()
					__UpperCamelCase				:  Any								=			os.path.join(snake_case		,  F'''{weight_name}.dat'''							)
					if index is not None:
										if dtype is None:
															__UpperCamelCase				:  Dict								=			str(array.dtype							)
										__UpperCamelCase				:  List[Any]								=			{"""dtype""": dtype, """shape""": list(array.shape							)}
					if array.ndim == 0:
										__UpperCamelCase				:  int								=			array[None]
					__UpperCamelCase				:  Optional[Any]								=			np.memmap(snake_case		,  dtype=array.dtype		,  mode="""w+"""		,  shape=array.shape							)
					__UpperCamelCase				:  str								=			array[:]
					file_array.flush()
					return index





def 							A__   (snake_case      :		Optional[Any]		,  snake_case      :		str							)    ->	List[Any]:
					__UpperCamelCase				:  Optional[Any]								=			tuple(weight_info["""shape"""]							)
					if shape == ():
										# NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
										__UpperCamelCase				:  Tuple								=			(1,)

					__UpperCamelCase				:  List[Any]								=			weight_info["""dtype"""]
					if dtype == "bfloat16":
										# NumPy does not support bfloat16 so this was saved as a int16
										__UpperCamelCase				:  List[str]								=			"""int16"""

					__UpperCamelCase				:  List[str]								=			np.memmap(snake_case		,  dtype=snake_case		,  shape=snake_case		,  mode="""r"""							)

					if len(weight_info["""shape"""]							) == 0:
										__UpperCamelCase				:  int								=			weight[0]
					__UpperCamelCase				:  Any								=			torch.tensor(snake_case							)
					if weight_info["dtype"] == "bfloat16":
										__UpperCamelCase				:  int								=			weight.view(torch.bfloataa							)

					return weight





def 							A__   (snake_case      :		Dict		,  snake_case      :		List[str]							)    ->	Optional[Any]:
					if index is None or len(snake_case							) == 0:
										# Nothing to save
										return

					__UpperCamelCase				:  Dict								=			os.path.join(snake_case		,  """index.json"""							)
					if os.path.isfile(snake_case							):
										with open(snake_case		,  """r"""		,  encoding="""utf-8"""							) as f:
															__UpperCamelCase				:  Any								=			json.load(snake_case							)
					else:
										__UpperCamelCase				:  Union[str, Any]								=			{}
					current_index.update(snake_case							)

					with open(snake_case		,  """w"""		,  encoding="""utf-8"""							) as f:
										json.dump(snake_case		,  snake_case		,  indent=2							)





def 							A__   (snake_case      :		Union[str, os.PathLike]		,  snake_case      :		Dict[str, torch.Tensor]							)    ->	str:
					os.makedirs(snake_case		,  exist_ok=snake_case							)
					__UpperCamelCase				:  Dict								=			{}
					for name, parameter in state_dict.items():
										__UpperCamelCase				:  int								=			offload_weight(snake_case		,  snake_case		,  snake_case		,  index=snake_case							)

					# Update index
					save_offload_index(snake_case		,  snake_case							)


class       SCREAMING_SNAKE_CASE_							(					_UpperCamelCase      ):



	"""simple docstring"""


	def __init__(				self		:	Any				,    lowerCAmelCase		:	Mapping				,    lowerCAmelCase		:	str  )  ->					Tuple:







						"""simple docstring"""







						__UpperCamelCase				:  Tuple								=			dataset
						__UpperCamelCase				:  List[str]								=			prefix






	def __getitem__(				self		:	List[Any]				,    lowerCAmelCase		:	List[str]  )  ->					Tuple:







						"""simple docstring"""







						return self.dataset[F'''{self.prefix}{key}''']






	def __iter__(				self		:	List[Any]  )  ->					List[Any]:







						"""simple docstring"""







						return iter([key for key in self.dataset if key.startswith(self.prefix  )]  )






	def __len__(				self		:	List[str]  )  ->					Tuple:







						"""simple docstring"""







						return len(self.dataset  )






class       SCREAMING_SNAKE_CASE_							(					_UpperCamelCase      ):



	"""simple docstring"""


	def __init__(				self		:	List[str]				,    lowerCAmelCase		:	Dict[str, torch.Tensor] = None				,    lowerCAmelCase		:	Optional[Union[str, os.PathLike]] = None				,    lowerCAmelCase		:	Mapping = None				,    lowerCAmelCase		:	Tuple=None				,    )  ->					Any:







						"""simple docstring"""







						if state_dict is None and save_folder is None:
											raise ValueError("""Need either a `state_dict` or a `save_folder` containing offloaded weights."""  )

						__UpperCamelCase				:  Any								=			{} if state_dict is None else state_dict
						__UpperCamelCase				:  Dict								=			save_folder
						if index is None and save_folder is not None:
											with open(os.path.join(lowerCAmelCase				,    """index.json"""  )  ) as f:
																__UpperCamelCase				:  str								=			json.load(lowerCAmelCase  )
						__UpperCamelCase				:  Optional[Any]								=			{} if index is None else index
						__UpperCamelCase				:  Dict								=			list(self.state_dict.keys()  )
						self.all_keys.extend([key for key in self.index if key not in self.all_keys]  )
						__UpperCamelCase				:  Dict								=			device






	def __getitem__(				self		:	Tuple				,    lowerCAmelCase		:	str  )  ->					List[str]:







						"""simple docstring"""







						if key in self.state_dict:
											return self.state_dict[key]
						__UpperCamelCase				:  int								=			self.index[key]
						if weight_info.get("""safetensors_file"""  ) is not None:
											if not is_safetensors_available():
																raise ImportError("""These offloaded weights require the use of safetensors: `pip install safetensors`."""  )

											from safetensors import safe_open

											__UpperCamelCase				:  Optional[int]								=			"""cpu""" if self.device is None else self.device
											with safe_open(weight_info["""safetensors_file"""]				,    framework="""pt"""				,    device=lowerCAmelCase  ) as f:
																__UpperCamelCase				:  Union[str, Any]								=			f.get_tensor(weight_info.get("""weight_name"""				,    lowerCAmelCase  )  )

											if "dtype" in weight_info:
																return tensor.to(getattr(lowerCAmelCase				,    weight_info["""dtype"""]  )  )
											else:
																return tensor

						__UpperCamelCase				:  int								=			os.path.join(self.save_folder				,    F'''{key}.dat'''  )
						return load_offloaded_weight(lowerCAmelCase				,    lowerCAmelCase  )






	def __iter__(				self		:	List[str]  )  ->					Union[str, Any]:







						"""simple docstring"""







						return iter(self.all_keys  )






	def __len__(				self		:	Union[str, Any]  )  ->					List[str]:







						"""simple docstring"""







						return len(self.all_keys  )






def 							A__   (snake_case      :		Dict[str, torch.Tensor]		,  snake_case      :		List[str]							)    ->	Union[str, Any]:
					__UpperCamelCase				:  Dict								=			{}
					for module_name in submodule_names:
										# We want to catch module_name parameter (module_name.xxx) or potentially module_name, but not any of the
										# submodules that could being like module_name (transformers.h.1 and transformers.h.10 for instance)
										result.update(
										    {
										        key: param
										        for key, param in state_dict.items()
										        if key == module_name or key.startswith(module_name + """."""							)
										    }							)
					return result












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)






