

import argparse
import os

import torch
from torchvision.datasets.utils import download_url

from diffusers import AutoencoderKL, DDIMScheduler, DiTPipeline, TransformeraDModel


a_       :Dict				  =				{512: "DiT-XL-2-512x512.pt", 256: "DiT-XL-2-256x256.pt"}
def 	lowercase_							(A			:			Dict				):
       snake_case__							:     str            =     F'''pretrained_models/{model_name}'''
       if not os.path.isfile(A				):
              os.makedirs('pretrained_models'		, exist_ok=A				)
              snake_case__							:     int            =     F'''https://dl.fbaipublicfiles.com/DiT/models/{model_name}'''
              download_url(A		, 'pretrained_models'				)
       snake_case__							:     Dict            =     torch.load(A		, map_location=lambda A		, A				: storage				)
       return model
def 	lowercase_							(A			:			Union[str, Any]				):
       snake_case__							:     Optional[Any]            =     download_model(pretrained_models[args.image_size]				)

       snake_case__							:     Optional[int]            =     state_dict['x_embedder.proj.weight']
       snake_case__							:     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(2_8				):
              snake_case__							:     str            =     state_dict[
                  't_embedder.mlp.0.weight'
              ]
              snake_case__							:     Tuple            =     state_dict[
                  't_embedder.mlp.0.bias'
              ]
              snake_case__							:     Optional[Any]            =     state_dict[
                  't_embedder.mlp.2.weight'
              ]
              snake_case__							:     List[str]            =     state_dict[
                  't_embedder.mlp.2.bias'
              ]
              snake_case__							:     List[Any]            =     state_dict[
                  'y_embedder.embedding_table.weight'
              ]

              snake_case__							:     Optional[int]            =     state_dict[
                  F'''blocks.{depth}.adaLN_modulation.1.weight'''
              ]
              snake_case__							:     List[str]            =     state_dict[
                  F'''blocks.{depth}.adaLN_modulation.1.bias'''
              ]

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

              snake_case__							:     Optional[Any]            =     q
              snake_case__							:     Union[str, Any]            =     q_bias
              snake_case__							:     Optional[Any]            =     k
              snake_case__							:     Optional[int]            =     k_bias
              snake_case__							:     int            =     v
              snake_case__							:     str            =     v_bias

              snake_case__							:     int            =     state_dict[
                  F'''blocks.{depth}.attn.proj.weight'''
              ]
              snake_case__							:     Tuple            =     state_dict[F'''blocks.{depth}.attn.proj.bias''']

              snake_case__							:     Optional[int]            =     state_dict[F'''blocks.{depth}.mlp.fc1.weight''']
              snake_case__							:     Optional[Any]            =     state_dict[F'''blocks.{depth}.mlp.fc1.bias''']
              snake_case__							:     int            =     state_dict[F'''blocks.{depth}.mlp.fc2.weight''']
              snake_case__							:     str            =     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'				)

       snake_case__							:     Optional[int]            =     state_dict['final_layer.adaLN_modulation.1.weight']
       snake_case__							:     str            =     state_dict['final_layer.adaLN_modulation.1.bias']
       snake_case__							:     str            =     state_dict['final_layer.linear.weight']
       snake_case__							:     Optional[int]            =     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
       snake_case__							:     Optional[Any]            =     TransformeraDModel(
           sample_size=args.image_size // 8		, num_layers=2_8		, attention_head_dim=7_2		, in_channels=4		, out_channels=8		, patch_size=2		, attention_bias=A		, num_attention_heads=1_6		, activation_fn='gelu-approximate'		, num_embeds_ada_norm=1_0_0_0		, norm_type='ada_norm_zero'		, norm_elementwise_affine=A		, )
       transformer.load_state_dict(A		, strict=A				)

       snake_case__							:     Dict            =     DDIMScheduler(
           num_train_timesteps=1_0_0_0		, beta_schedule='linear'		, prediction_type='epsilon'		, clip_sample=A		, )

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

       snake_case__							:     Optional[int]            =     DiTPipeline(transformer=A		, vae=A		, scheduler=A				)

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


if __name__ == "__main__":
 a_       :Optional[Any]				  =				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_       :int				  =				parser.parse_args()
 main(args)











import shutil
import tempfile
import unittest

import pytest

from transformers import WhisperTokenizer, is_speech_available
from transformers.testing_utils import require_sentencepiece, require_torch, require_torchaudio

from .test_feature_extraction_whisper import floats_list


if is_speech_available():
 from transformers import WhisperFeatureExtractor, WhisperProcessor


a_       :List[Any]				  =				50_358
a_       :int				  =				50_362




@require_torch
@require_torchaudio
@require_sentencepiece
class  snake_case__						(	unittest.TestCase					):






     """simple docstring"""


     def        lowercase_       (	self  :  Dict )       ->List[Any]:
            snake_case__							:     int            =     'openai/whisper-small.en'
            snake_case__							:     List[Any]            =     tempfile.mkdtemp()






     def        lowercase_       (	self  :  Any,			**_snake_case  :  List[str] )       ->Union[str, Any]:
            return WhisperTokenizer.from_pretrained(self.checkpoint,			**_snake_case )






     def        lowercase_       (	self  :  Union[str, Any],			**_snake_case  :  int )       ->Optional[int]:
            return WhisperFeatureExtractor.from_pretrained(self.checkpoint,			**_snake_case )






     def        lowercase_       (	self  :  Any )       ->Union[str, Any]:
            shutil.rmtree(self.tmpdirname )






     def        lowercase_       (	self  :  Dict )       ->Tuple:
            snake_case__							:     Dict            =     self.get_tokenizer()
            snake_case__							:     List[str]            =     self.get_feature_extractor()

            snake_case__							:     List[Any]            =     WhisperProcessor(tokenizer=_snake_case,			feature_extractor=_snake_case )

            processor.save_pretrained(self.tmpdirname )
            snake_case__							:     Dict            =     WhisperProcessor.from_pretrained(self.tmpdirname )

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

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






     def        lowercase_       (	self  :  List[Any] )       ->str:
            snake_case__							:     str            =     WhisperProcessor(tokenizer=self.get_tokenizer(),			feature_extractor=self.get_feature_extractor() )
            processor.save_pretrained(self.tmpdirname )

            snake_case__							:     Dict            =     self.get_tokenizer(bos_token='(BOS)',			eos_token='(EOS)' )
            snake_case__							:     List[str]            =     self.get_feature_extractor(do_normalize=_snake_case,			padding_value=1.0 )

            snake_case__							:     List[Any]            =     WhisperProcessor.from_pretrained(
                self.tmpdirname,			bos_token='(BOS)',			eos_token='(EOS)',			do_normalize=_snake_case,			padding_value=1.0 )

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

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






     def        lowercase_       (	self  :  Union[str, Any] )       ->Dict:
            snake_case__							:     Dict            =     self.get_feature_extractor()
            snake_case__							:     Dict            =     self.get_tokenizer()

            snake_case__							:     List[Any]            =     WhisperProcessor(tokenizer=_snake_case,			feature_extractor=_snake_case )

            snake_case__							:     int            =     floats_list((3, 1_0_0_0) )

            snake_case__							:     Optional[int]            =     feature_extractor(_snake_case,			return_tensors='np' )
            snake_case__							:     Optional[int]            =     processor(_snake_case,			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        lowercase_       (	self  :  Dict )       ->Any:
            snake_case__							:     Tuple            =     self.get_feature_extractor()
            snake_case__							:     Optional[int]            =     self.get_tokenizer()

            snake_case__							:     Optional[int]            =     WhisperProcessor(tokenizer=_snake_case,			feature_extractor=_snake_case )

            snake_case__							:     Any            =     'This is a test string'

            snake_case__							:     Dict            =     processor(text=_snake_case )

            snake_case__							:     List[str]            =     tokenizer(_snake_case )

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






     def        lowercase_       (	self  :  Dict )       ->Dict:
            snake_case__							:     Optional[Any]            =     self.get_feature_extractor()
            snake_case__							:     Optional[int]            =     self.get_tokenizer()

            snake_case__							:     Optional[int]            =     WhisperProcessor(tokenizer=_snake_case,			feature_extractor=_snake_case )

            snake_case__							:     Union[str, Any]            =     [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

            snake_case__							:     List[str]            =     processor.batch_decode(_snake_case )
            snake_case__							:     Dict            =     tokenizer.batch_decode(_snake_case )

            self.assertListEqual(_snake_case,			_snake_case )






     def        lowercase_       (	self  :  Union[str, Any] )       ->Optional[Any]:
            snake_case__							:     str            =     self.get_feature_extractor()
            snake_case__							:     Dict            =     self.get_tokenizer()

            snake_case__							:     Any            =     WhisperProcessor(tokenizer=_snake_case,			feature_extractor=_snake_case )

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






     def        lowercase_       (	self  :  int )       ->Dict:
            snake_case__							:     Tuple            =     self.get_feature_extractor()
            snake_case__							:     Tuple            =     self.get_tokenizer()

            snake_case__							:     Optional[int]            =     WhisperProcessor(tokenizer=_snake_case,			feature_extractor=_snake_case )
            snake_case__							:     Any            =     processor.get_decoder_prompt_ids(task='transcribe',			no_timestamps=_snake_case )

            self.assertIsInstance(_snake_case,			_snake_case )
            for ids in forced_decoder_ids:
                   self.assertIsInstance(_snake_case,			(list, tuple) )

            snake_case__							:     List[str]            =     [TRANSCRIBE, NOTIMESTAMPS]
            self.assertListEqual([ids[-1] for ids in forced_decoder_ids],			_snake_case )






     def        lowercase_       (	self  :  Optional[int] )       ->int:
            snake_case__							:     List[Any]            =     WhisperProcessor(tokenizer=self.get_tokenizer(),			feature_extractor=self.get_feature_extractor() )
            snake_case__							:     Optional[Any]            =     processor.get_prompt_ids('Mr. Quilter' )
            snake_case__							:     List[Any]            =     processor.tokenizer.decode(_snake_case )

            self.assertListEqual(prompt_ids.tolist(),			[5_0_3_6_0, 1_7_7_0, 1_3, 2_2_6_4, 3_4_6, 3_5_3] )
            self.assertEqual(_snake_case,			'<|startofprev|> Mr. Quilter' )






     def        lowercase_       (	self  :  Tuple )       ->Tuple:
            snake_case__							:     int            =     WhisperProcessor(tokenizer=self.get_tokenizer(),			feature_extractor=self.get_feature_extractor() )
            snake_case__							:     List[str]            =     processor.get_prompt_ids('' )
            snake_case__							:     List[Any]            =     processor.tokenizer.decode(_snake_case )

            self.assertListEqual(prompt_ids.tolist(),			[5_0_3_6_0, 2_2_0] )
            self.assertEqual(_snake_case,			'<|startofprev|> ' )






     def        lowercase_       (	self  :  Tuple )       ->str:
            snake_case__							:     Any            =     WhisperProcessor(tokenizer=self.get_tokenizer(),			feature_extractor=self.get_feature_extractor() )

            def _test_prompt_error_raised_helper(_snake_case  :  str,			_snake_case  :  int ):
                   with pytest.raises(_snake_case ) as excinfo:
                          processor.get_prompt_ids(_snake_case )
                   snake_case__							:     Dict            =     F'''Encountered text in the prompt corresponding to disallowed special token: {special_token}.'''
                   self.assertEqual(_snake_case,			str(excinfo.value ) )

            _test_prompt_error_raised_helper('<|startofprev|> test',			'<|startofprev|>' )
            _test_prompt_error_raised_helper('test <|notimestamps|>',			'<|notimestamps|>' )
            _test_prompt_error_raised_helper('test <|zh|> test <|transcribe|>',			'<|zh|>' )






