






'''simple docstring'''





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 			__UpperCAmelCase				(lowercase__				)		->				Optional[int]:

						'''simple docstring'''


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



def 			__UpperCAmelCase				(lowercase__				)		->				str:

						'''simple docstring'''


						a_				     =   download_model(pretrained_models[args.image_size]				)

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

												a_				     =   state_dict[
												    F"""blocks.{depth}.adaLN_modulation.1.weight"""
												]
												a_				     =   state_dict[
												    F"""blocks.{depth}.adaLN_modulation.1.bias"""
												]

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

												a_				     =   q
												a_				     =   q_bias
												a_				     =   k
												a_				     =   k_bias
												a_				     =   v
												a_				     =   v_bias

												a_				     =   state_dict[
												    F"""blocks.{depth}.attn.proj.weight"""
												]
												a_				     =   state_dict[F"""blocks.{depth}.attn.proj.bias"""]

												a_				     =   state_dict[F"""blocks.{depth}.mlp.fc1.weight"""]
												a_				     =   state_dict[F"""blocks.{depth}.mlp.fc1.bias"""]
												a_				     =   state_dict[F"""blocks.{depth}.mlp.fc2.weight"""]
												a_				     =   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"				)

						a_				     =   state_dict["final_layer.adaLN_modulation.1.weight"]
						a_				     =   state_dict["final_layer.adaLN_modulation.1.bias"]
						a_				     =   state_dict["final_layer.linear.weight"]
						a_				     =   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
						a_				     =   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=lowercase__	,num_attention_heads=16	,activation_fn="gelu-approximate"	,num_embeds_ada_norm=1000	,norm_type="ada_norm_zero"	,norm_elementwise_affine=lowercase__	,)
						transformer.load_state_dict(lowercase__	,strict=lowercase__				)

						a_				     =   DDIMScheduler(
						    num_train_timesteps=1000	,beta_schedule="linear"	,prediction_type="epsilon"	,clip_sample=lowercase__	,)

						a_				     =   AutoencoderKL.from_pretrained(args.vae_model				)

						a_				     =   DiTPipeline(transformer=lowercase__	,vae=lowercase__	,scheduler=lowercase__				)

						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)











'''simple docstring'''





import json
import os
import shutil
import tempfile
import unittest

from transformers import is_speech_available, is_torch_available
from transformers.models.speechta import SpeechTaTokenizer
from transformers.testing_utils import get_tests_dir, require_torch
from transformers.utils import FEATURE_EXTRACTOR_NAME


if is_speech_available() and is_torch_available():
							from transformers import SpeechTaFeatureExtractor, SpeechTaProcessor

							from .test_feature_extraction_speechta import floats_list


a_      = get_tests_dir('fixtures/test_sentencepiece_bpe_char.model')





@require_torch
class 		SCREAMING_SNAKE_CASE__   (     unittest.TestCase    ):



		def _lowerCAmelCase     (    self:    str)  ->Optional[Any]:






								'''simple docstring'''


								a_				     =   tempfile.mkdtemp()

								a_				     =   SpeechTaTokenizer(a)
								tokenizer.save_pretrained(self.tmpdirname)

								a_				     =   {
								    "feature_size": 1,
								    "padding_value": 0.0,
								    "sampling_rate": 1_60_00,
								    "do_normalize": False,
								    "num_mel_bins": 80,
								    "hop_length": 16,
								    "win_length": 64,
								    "win_function": "hann_window",
								    "fmin": 80,
								    "fmax": 76_00,
								    "mel_floor": 1e-10,
								    "reduction_factor": 2,
								    "return_attention_mask": True,
								}

								a_				     =   os.path.join(self.tmpdirname				,   a)
								with open(self.feature_extraction_file				,   "w"				,   encoding="utf-8") as fp:
														fp.write(json.dumps(a) + "\n")



		def _lowerCAmelCase     (    self:    Any				,   **a:    List[Any])  ->Optional[Any]:






								'''simple docstring'''


								return SpeechTaTokenizer.from_pretrained(self.tmpdirname				,   **a)



		def _lowerCAmelCase     (    self:    Union[str, Any]				,   **a:    int)  ->str:






								'''simple docstring'''


								return SpeechTaFeatureExtractor.from_pretrained(self.tmpdirname				,   **a)



		def _lowerCAmelCase     (    self:    int)  ->Optional[Any]:






								'''simple docstring'''


								shutil.rmtree(self.tmpdirname)



		def _lowerCAmelCase     (    self:    Any)  ->str:






								'''simple docstring'''


								a_				     =   self.get_tokenizer()
								a_				     =   self.get_feature_extractor()

								a_				     =   SpeechTaProcessor(tokenizer=a				,   feature_extractor=a)

								processor.save_pretrained(self.tmpdirname)
								a_				     =   SpeechTaProcessor.from_pretrained(self.tmpdirname)

								self.assertEqual(processor.tokenizer.get_vocab()				,   tokenizer.get_vocab())
								self.assertIsInstance(processor.tokenizer				,   a)

								self.assertEqual(processor.feature_extractor.to_json_string()				,   feature_extractor.to_json_string())
								self.assertIsInstance(processor.feature_extractor				,   a)



		def _lowerCAmelCase     (    self:    List[Any])  ->Optional[int]:






								'''simple docstring'''


								a_				     =   SpeechTaProcessor(tokenizer=self.get_tokenizer()				,   feature_extractor=self.get_feature_extractor())
								processor.save_pretrained(self.tmpdirname)

								a_				     =   self.get_tokenizer(bos_token="(BOS)"				,   eos_token="(EOS)")
								a_				     =   self.get_feature_extractor(do_normalize=a				,   padding_value=1.0)

								a_				     =   SpeechTaProcessor.from_pretrained(
								    self.tmpdirname				,   bos_token="(BOS)"				,   eos_token="(EOS)"				,   do_normalize=a				,   padding_value=1.0)

								self.assertEqual(processor.tokenizer.get_vocab()				,   tokenizer_add_kwargs.get_vocab())
								self.assertIsInstance(processor.tokenizer				,   a)

								self.assertEqual(processor.feature_extractor.to_json_string()				,   feature_extractor_add_kwargs.to_json_string())
								self.assertIsInstance(processor.feature_extractor				,   a)



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






								'''simple docstring'''


								a_				     =   self.get_feature_extractor()
								a_				     =   self.get_tokenizer()

								a_				     =   SpeechTaProcessor(tokenizer=a				,   feature_extractor=a)

								a_				     =   floats_list((3, 10_00))

								a_				     =   feature_extractor(audio=a				,   return_tensors="np")
								a_				     =   processor(audio=a				,   return_tensors="np")

								for key in input_feat_extract.keys():
														self.assertAlmostEqual(input_feat_extract[key].sum()				,   input_processor[key].sum()				,   delta=1e-2)



		def _lowerCAmelCase     (    self:    Tuple)  ->Union[str, Any]:






								'''simple docstring'''


								a_				     =   self.get_feature_extractor()
								a_				     =   self.get_tokenizer()

								a_				     =   SpeechTaProcessor(tokenizer=a				,   feature_extractor=a)

								a_				     =   floats_list((3, 10_00))

								a_				     =   feature_extractor(audio_target=a				,   return_tensors="np")
								a_				     =   processor(audio_target=a				,   return_tensors="np")

								for key in input_feat_extract.keys():
														self.assertAlmostEqual(input_feat_extract[key].sum()				,   input_processor[key].sum()				,   delta=1e-2)



		def _lowerCAmelCase     (    self:    str)  ->Dict:






								'''simple docstring'''


								a_				     =   self.get_feature_extractor()
								a_				     =   self.get_tokenizer()

								a_				     =   SpeechTaProcessor(tokenizer=a				,   feature_extractor=a)

								a_				     =   "This is a test string"

								a_				     =   processor(text=a)
								a_				     =   tokenizer(a)

								for key in encoded_tok.keys():
														self.assertListEqual(encoded_tok[key]				,   encoded_processor[key])



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






								'''simple docstring'''


								a_				     =   self.get_feature_extractor()
								a_				     =   self.get_tokenizer()

								a_				     =   SpeechTaProcessor(tokenizer=a				,   feature_extractor=a)

								a_				     =   "This is a test string"

								a_				     =   processor(text_target=a)
								a_				     =   tokenizer(a)

								for key in encoded_tok.keys():
														self.assertListEqual(encoded_tok[key]				,   encoded_processor[key])



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






								'''simple docstring'''


								a_				     =   self.get_feature_extractor()
								a_				     =   self.get_tokenizer()

								a_				     =   SpeechTaProcessor(tokenizer=a				,   feature_extractor=a)

								a_				     =   [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

								a_				     =   processor.batch_decode(a)
								a_				     =   tokenizer.batch_decode(a)

								self.assertListEqual(a				,   a)




		def _lowerCAmelCase     (    self:    Optional[int])  ->Optional[int]:






								'''simple docstring'''


								a_				     =   self.get_feature_extractor()
								a_				     =   self.get_tokenizer()

								a_				     =   SpeechTaProcessor(tokenizer=a				,   feature_extractor=a)

								self.assertListEqual(
								    processor.model_input_names				,   feature_extractor.model_input_names				,   msg="`processor` and `feature_extractor` model input names do not match"				,   )



