



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__     ,    snake_case__     ,    snake_case__							:     int            =     torch.chunk(state_dict[F'''blocks.{depth}.attn.qkv.weight''']		, 3		, dim=0				)
              snake_case__     ,    snake_case__     ,    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)











from manim import *




class  snake_case__						(	lowerCAmelCase_					):






     """simple docstring"""


     def        lowercase_       (	self  :  List[Any] )       ->Dict:
            snake_case__							:     str            =     Rectangle(height=0.5,			width=0.5 )
            snake_case__							:     str            =     Rectangle(height=0.4_6,			width=0.4_6 ).set_stroke(width=0 )
            snake_case__							:     List[Any]            =     Rectangle(height=0.2_5,			width=0.2_5 )

            snake_case__							:     Any            =     [mem.copy() for i in range(6 )]
            snake_case__							:     Optional[Any]            =     [mem.copy() for i in range(6 )]
            snake_case__							:     int            =     VGroup(*_snake_case ).arrange(_snake_case,			buff=0 )
            snake_case__							:     Optional[int]            =     VGroup(*_snake_case ).arrange(_snake_case,			buff=0 )
            snake_case__							:     Any            =     VGroup(_snake_case,			_snake_case ).arrange(_snake_case,			buff=0 )
            snake_case__							:     Optional[Any]            =     Text('CPU',			font_size=2_4 )
            snake_case__							:     str            =     Group(_snake_case,			_snake_case ).arrange(_snake_case,			buff=0.5,			aligned_edge=_snake_case )
            cpu.move_to([-2.5, -0.5, 0] )
            self.add(_snake_case )

            snake_case__							:     Optional[int]            =     [mem.copy() for i in range(4 )]
            snake_case__							:     int            =     VGroup(*_snake_case ).arrange(_snake_case,			buff=0 )
            snake_case__							:     str            =     Text('GPU',			font_size=2_4 )
            snake_case__							:     List[Any]            =     Group(_snake_case,			_snake_case ).arrange(_snake_case,			buff=0.5,			aligned_edge=_snake_case )
            gpu.move_to([-1, -1, 0] )
            self.add(_snake_case )

            snake_case__							:     Optional[int]            =     [mem.copy() for i in range(6 )]
            snake_case__							:     Union[str, Any]            =     VGroup(*_snake_case ).arrange(_snake_case,			buff=0 )

            snake_case__							:     int            =     Text('Model',			font_size=2_4 )
            snake_case__							:     List[Any]            =     Group(_snake_case,			_snake_case ).arrange(_snake_case,			buff=0.5,			aligned_edge=_snake_case )
            model.move_to([3, -1.0, 0] )
            self.add(_snake_case )

            snake_case__							:     Any            =     []
            snake_case__							:     Optional[int]            =     []

            for i, rect in enumerate(_snake_case ):
                   rect.set_stroke(_snake_case )

                   snake_case__							:     Union[str, Any]            =     Rectangle(height=0.4_6 / 4,			width=0.4_6 / 3 ).set_stroke(width=0.0 ).set_fill(_snake_case,			opacity=0.7 )

                   if i == 0:
                          cpu_target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT ),			buff=0.0_2,			direction=_snake_case )
                          cpu_target.set_x(cpu_target.get_x() + 0.1 )
                   elif i == 3:
                          cpu_target.next_to(model_cpu_arr[0],			direction=_snake_case,			buff=0.0 )
                   else:
                          cpu_target.next_to(model_cpu_arr[i - 1],			direction=_snake_case,			buff=0.0 )
                   self.add(_snake_case )
                   model_cpu_arr.append(_snake_case )

            self.add(*_snake_case,			*_snake_case )

            snake_case__							:     Optional[Any]            =     [meta_mem.copy() for i in range(6 )]
            snake_case__							:     List[Any]            =     [meta_mem.copy() for i in range(6 )]
            snake_case__							:     Dict            =     VGroup(*_snake_case ).arrange(_snake_case,			buff=0 )
            snake_case__							:     Tuple            =     VGroup(*_snake_case ).arrange(_snake_case,			buff=0 )
            snake_case__							:     int            =     VGroup(_snake_case,			_snake_case ).arrange(_snake_case,			buff=0 )
            snake_case__							:     Union[str, Any]            =     Text('Disk',			font_size=2_4 )
            snake_case__							:     int            =     Group(_snake_case,			_snake_case ).arrange(_snake_case,			buff=0.5,			aligned_edge=_snake_case )
            disk.move_to([-4.0, -1.2_5, 0] )
            self.add(_snake_case,			_snake_case )

            snake_case__							:     str            =     []

            for i in range(6 ):
                   snake_case__							:     str            =     fill.copy().set_fill(_snake_case,			opacity=0.8 )
                   target.move_to(disk_left_col_base[i] ).scale(0.5 )
                   cpu_disk_arr.append(_snake_case )

            self.add(*_snake_case )

            snake_case__							:     int            =     Square(side_length=2.2 )
            key.move_to([-5, 2, 0] )

            snake_case__							:     int            =     MarkupText(
                F'''<b>Key:</b>\n\n<span fgcolor=\'{YELLOW}\'>●</span> Empty Model''',			font_size=1_8,			)

            key_text.move_to([-5, 2.4, 0] )

            self.add(_snake_case,			_snake_case )

            snake_case__							:     Optional[int]            =     MarkupText(
                F'''<span fgcolor=\'{BLUE}\'>●</span> Checkpoint''',			font_size=1_8,			)

            blue_text.next_to(_snake_case,			DOWN * 2.4,			aligned_edge=key_text.get_left() )
            self.add(_snake_case )

            snake_case__							:     Optional[int]            =     MarkupText(F'''The offloaded weights are all sent to the CPU.''',			font_size=2_4 )
            step_a.move_to([2, 2, 0] )

            self.play(Write(_snake_case,			run_time=3 ) )

            for i in range(6 ):
                   snake_case__							:     int            =     cpu_disk_arr[i]
                   snake_case__							:     Optional[Any]            =     rect.copy().set_fill(_snake_case,			opacity=0.8 ).scale(2.0 )
                   cpa.generate_target()
                   cpa.target.move_to(model_base[i] )

                   if i == 0:
                          rect.set_fill(_snake_case,			opacity=0.8 )
                          rect.generate_target()
                          rect.target.move_to(cpu_left_col_base[0] ).scale(2.0 )

                          self.remove(
                              *_snake_case,			*_snake_case,			)

                   else:
                          rect.generate_target()
                          rect.target.move_to(cpu_left_col_base[i] ).scale(2.0 )
                   self.play(MoveToTarget(_snake_case ),			MoveToTarget(_snake_case ),			model_base[i].animate.set_stroke(_snake_case ) )
            self.play(FadeOut(_snake_case ) )

            snake_case__							:     Optional[Any]            =     MarkupText(
                F'''Finally, hooks are added to each weight in the model\nto transfer the weights from CPU to GPU\n\t\tand back when needed.''',			font_size=2_4,			)
            step_a.move_to([2, 2, 0] )

            self.play(Write(_snake_case,			run_time=3 ) )

            snake_case__							:     str            =     []
            snake_case__							:     Optional[Any]            =     []
            for i in range(6 ):
                   snake_case__							:     List[Any]            =     Arrow(start=_snake_case,			end=_snake_case,			color=_snake_case,			buff=0.5 )
                   a.next_to(model_base[i].get_left(),			_snake_case,			buff=0.2 )
                   arrows.append(_snake_case )
                   animations.append(Write(_snake_case ) )
            self.play(*_snake_case )
            self.wait()






