





from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin






@dataclass
class 					lowercase__ (			__lowerCamelCase       ):

   '''simple docstring'''


   a		:							torch.FloatTensor






class 					lowercase__ (			__lowerCamelCase    ,       __lowerCamelCase       ):

   '''simple docstring'''


   @register_to_config
   def __init__(					self,	__magic_name__ = 32,	__magic_name__ = 64,	__magic_name__ = 20,	__magic_name__ = 768,	__magic_name__=77,	__magic_name__=4,	__magic_name__ = 0.0,	__magic_name__ = "silu",	__magic_name__ = None,	__magic_name__ = None,	__magic_name__ = "linear",	__magic_name__ = "prd",	__magic_name__ = None,	__magic_name__ = None,	__magic_name__ = None,	)				->   str:






       """simple docstring"""
       super().__init__()
       UpperCamelCase__							:       Optional[int]      =   num_attention_heads
       UpperCamelCase__							:       List[str]      =   attention_head_dim
       UpperCamelCase__							:       Union[str, Any]      =   num_attention_heads * attention_head_dim
       UpperCamelCase__							:       str      =   additional_embeddings

       UpperCamelCase__							:       Any      =   time_embed_dim or inner_dim
       UpperCamelCase__							:       str      =   embedding_proj_dim or embedding_dim
       UpperCamelCase__							:       Union[str, Any]      =   clip_embed_dim or embedding_dim

       UpperCamelCase__							:       Union[str, Any]      =   Timesteps(__magic_name__,	__magic_name__,	0							)
       UpperCamelCase__							:       Optional[Any]      =   TimestepEmbedding(__magic_name__,	__magic_name__,	out_dim=__magic_name__,	act_fn=__magic_name__							)

       UpperCamelCase__							:       Union[str, Any]      =   nn.Linear(__magic_name__,	__magic_name__							)

       if embedding_proj_norm_type is None:
           UpperCamelCase__							:       List[str]      =   None
       elif embedding_proj_norm_type == "layer":
           UpperCamelCase__							:       Tuple      =   nn.LayerNorm(__magic_name__							)
       else:
           raise ValueError(f"unsupported embedding_proj_norm_type: {embedding_proj_norm_type}"							)

       UpperCamelCase__							:       Dict      =   nn.Linear(__magic_name__,	__magic_name__							)

       if encoder_hid_proj_type is None:
           UpperCamelCase__							:       int      =   None
       elif encoder_hid_proj_type == "linear":
           UpperCamelCase__							:       Any      =   nn.Linear(__magic_name__,	__magic_name__							)
       else:
           raise ValueError(f"unsupported encoder_hid_proj_type: {encoder_hid_proj_type}"							)

       UpperCamelCase__							:       List[Any]      =   nn.Parameter(torch.zeros(1,	num_embeddings + additional_embeddings,	__magic_name__							)							)

       if added_emb_type == "prd":
           UpperCamelCase__							:       List[str]      =   nn.Parameter(torch.zeros(1,	1,	__magic_name__							)							)
       elif added_emb_type is None:
           UpperCamelCase__							:       Optional[int]      =   None
       else:
           raise ValueError(
               f"`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `'prd'` or `None`."							)

       UpperCamelCase__							:       Union[str, Any]      =   nn.ModuleList(
           [
               BasicTransformerBlock(
                   __magic_name__,	__magic_name__,	__magic_name__,	dropout=__magic_name__,	activation_fn='''gelu''',	attention_bias=__magic_name__,	)
               for d in range(__magic_name__							)
           ]							)

       if norm_in_type == "layer":
           UpperCamelCase__							:       str      =   nn.LayerNorm(__magic_name__							)
       elif norm_in_type is None:
           UpperCamelCase__							:       str      =   None
       else:
           raise ValueError(f"Unsupported norm_in_type: {norm_in_type}."							)

       UpperCamelCase__							:       Optional[Any]      =   nn.LayerNorm(__magic_name__							)

       UpperCamelCase__							:       int      =   nn.Linear(__magic_name__,	__magic_name__							)

       UpperCamelCase__							:       Optional[Any]      =   torch.full(
           [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings],	-1_0000.0							)
       causal_attention_mask.triu_(1							)
       UpperCamelCase__							:       List[Any]      =   causal_attention_mask[None, ...]
       self.register_buffer('''causal_attention_mask''',	__magic_name__,	persistent=__magic_name__							)

       UpperCamelCase__							:       Union[str, Any]      =   nn.Parameter(torch.zeros(1,	__magic_name__							)							)
       UpperCamelCase__							:       List[str]      =   nn.Parameter(torch.zeros(1,	__magic_name__							)							)




   @property
   # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
   def    UpperCamelCase__						(					self							)				->   Dict[str, AttentionProcessor]:






       """simple docstring"""
       UpperCamelCase__							:       Union[str, Any]      =   {}

       def fn_recursive_add_processors(__magic_name__,	__magic_name__,	__magic_name__							):
           if hasattr(__magic_name__,	'''set_processor'''							):
               UpperCamelCase__							:       Optional[int]      =   module.processor

           for sub_name, child in module.named_children():
               fn_recursive_add_processors(f"{name}.{sub_name}",	__magic_name__,	__magic_name__							)

           return processors

       for name, module in self.named_children():
           fn_recursive_add_processors(__magic_name__,	__magic_name__,	__magic_name__							)

       return processors




   def    UpperCamelCase__						(					self,	__magic_name__							)				->   Tuple:






       """simple docstring"""
       UpperCamelCase__							:       Tuple      =   len(self.attn_processors.keys()							)

       if isinstance(__magic_name__,	__magic_name__							) and len(__magic_name__							) != count:
           raise ValueError(
               f"A dict of processors was passed, but the number of processors {len(__magic_name__							)} does not match the"
               f" number of attention layers: {count}. Please make sure to pass {count} processor classes."							)

       def fn_recursive_attn_processor(__magic_name__,	__magic_name__,	__magic_name__							):
           if hasattr(__magic_name__,	'''set_processor'''							):
               if not isinstance(__magic_name__,	__magic_name__							):
                   module.set_processor(__magic_name__							)
               else:
                   module.set_processor(processor.pop(f"{name}.processor"							)							)

           for sub_name, child in module.named_children():
               fn_recursive_attn_processor(f"{name}.{sub_name}",	__magic_name__,	__magic_name__							)

       for name, module in self.named_children():
           fn_recursive_attn_processor(__magic_name__,	__magic_name__,	__magic_name__							)




   def    UpperCamelCase__						(					self							)				->   List[str]:






       """simple docstring"""
       self.set_attn_processor(AttnProcessor()							)




   def    UpperCamelCase__						(					self,	__magic_name__,	__magic_name__,	__magic_name__,	__magic_name__ = None,	__magic_name__ = None,	__magic_name__ = True,	)				->   List[Any]:






       """simple docstring"""
       UpperCamelCase__							:       Optional[int]      =   hidden_states.shape[0]

       UpperCamelCase__							:       Optional[Any]      =   timestep
       if not torch.is_tensor(__magic_name__							):
           UpperCamelCase__							:       Tuple      =   torch.tensor([timesteps],	dtype=torch.long,	device=hidden_states.device							)
       elif torch.is_tensor(__magic_name__							) and len(timesteps.shape							) == 0:
           UpperCamelCase__							:       List[str]      =   timesteps[None].to(hidden_states.device							)

       # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
       UpperCamelCase__							:       int      =   timesteps * torch.ones(__magic_name__,	dtype=timesteps.dtype,	device=timesteps.device							)

       UpperCamelCase__							:       Optional[Any]      =   self.time_proj(__magic_name__							)

       # timesteps does not contain any weights and will always return f32 tensors
       # but time_embedding might be fp16, so we need to cast here.
       UpperCamelCase__							:       List[Any]      =   timesteps_projected.to(dtype=self.dtype							)
       UpperCamelCase__							:       Union[str, Any]      =   self.time_embedding(__magic_name__							)

       if self.embedding_proj_norm is not None:
           UpperCamelCase__							:       List[Any]      =   self.embedding_proj_norm(__magic_name__							)

       UpperCamelCase__							:       int      =   self.embedding_proj(__magic_name__							)
       if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
           UpperCamelCase__							:       Tuple      =   self.encoder_hidden_states_proj(__magic_name__							)
       elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
           raise ValueError('''`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set'''							)

       UpperCamelCase__							:       Tuple      =   self.proj_in(__magic_name__							)

       UpperCamelCase__							:       List[Any]      =   self.positional_embedding.to(hidden_states.dtype							)

       UpperCamelCase__							:       str      =   []
       UpperCamelCase__							:       Any      =   0

       if encoder_hidden_states is not None:
           additional_embeds.append(__magic_name__							)
           additional_embeddings_len += encoder_hidden_states.shape[1]

       if len(proj_embeddings.shape							) == 2:
           UpperCamelCase__							:       Dict      =   proj_embeddings[:, None, :]

       if len(hidden_states.shape							) == 2:
           UpperCamelCase__							:       Optional[int]      =   hidden_states[:, None, :]

       UpperCamelCase__							:       List[Any]      =   additional_embeds + [
           proj_embeddings,
           time_embeddings[:, None, :],
           hidden_states,
       ]

       if self.prd_embedding is not None:
           UpperCamelCase__							:       str      =   self.prd_embedding.to(hidden_states.dtype							).expand(__magic_name__,	-1,	-1							)
           additional_embeds.append(__magic_name__							)

       UpperCamelCase__							:       Dict      =   torch.cat(
           __magic_name__,	dim=1,	)

       # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
       UpperCamelCase__							:       int      =   additional_embeddings_len + proj_embeddings.shape[1] + 1
       if positional_embeddings.shape[1] < hidden_states.shape[1]:
           UpperCamelCase__							:       Union[str, Any]      =   F.pad(
               __magic_name__,	(
                   0,
                   0,
                   additional_embeddings_len,
                   self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
               ),	value=0.0,	)

       UpperCamelCase__							:       Union[str, Any]      =   hidden_states + positional_embeddings

       if attention_mask is not None:
           UpperCamelCase__							:       List[str]      =   (1 - attention_mask.to(hidden_states.dtype							)) * -1_0000.0
           UpperCamelCase__							:       Union[str, Any]      =   F.pad(__magic_name__,	(0, self.additional_embeddings),	value=0.0							)
           UpperCamelCase__							:       str      =   (attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype							)
           UpperCamelCase__							:       Optional[int]      =   attention_mask.repeat_interleave(self.config.num_attention_heads,	dim=0							)

       if self.norm_in is not None:
           UpperCamelCase__							:       Optional[int]      =   self.norm_in(__magic_name__							)

       for block in self.transformer_blocks:
           UpperCamelCase__							:       Optional[int]      =   block(__magic_name__,	attention_mask=__magic_name__							)

       UpperCamelCase__							:       List[Any]      =   self.norm_out(__magic_name__							)

       if self.prd_embedding is not None:
           UpperCamelCase__							:       Union[str, Any]      =   hidden_states[:, -1]
       else:
           UpperCamelCase__							:       Optional[Any]      =   hidden_states[:, additional_embeddings_len:]

       UpperCamelCase__							:       Optional[int]      =   self.proj_to_clip_embeddings(__magic_name__							)

       if not return_dict:
           return (predicted_image_embedding,)

       return PriorTransformerOutput(predicted_image_embedding=__magic_name__							)




   def    UpperCamelCase__						(					self,	__magic_name__							)				->   List[str]:






       """simple docstring"""
       UpperCamelCase__							:       Dict      =   (prior_latents * self.clip_std) + self.clip_mean
       return prior_latents








from manim import *






class 					lowercase__ (			__lowerCamelCase       ):

   '''simple docstring'''


   def    UpperCamelCase__						(					self							)				->   List[str]:






       """simple docstring"""
       UpperCamelCase__							:       Any      =   Rectangle(height=0.5,	width=0.5							)
       UpperCamelCase__							:       Any      =   Rectangle(height=0.25,	width=0.25							)
       UpperCamelCase__							:       Dict      =   Rectangle(height=0.46,	width=0.46							).set_stroke(width=0							)

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

       UpperCamelCase__							:       int      =   [mem.copy() for i in range(4							)]
       UpperCamelCase__							:       str      =   VGroup(*__magic_name__							).arrange(__magic_name__,	buff=0							)
       UpperCamelCase__							:       Tuple      =   Text('''GPU''',	font_size=24							)
       UpperCamelCase__							:       str      =   Group(__magic_name__,	__magic_name__							).arrange(__magic_name__,	buff=0.5,	aligned_edge=__magic_name__							)
       gpu.move_to([-1, -1, 0]							)
       self.add(__magic_name__							)

       UpperCamelCase__							:       str      =   [mem.copy() for i in range(6							)]
       UpperCamelCase__							:       Tuple      =   VGroup(*__magic_name__							).arrange(__magic_name__,	buff=0							)

       UpperCamelCase__							:       Tuple      =   Text('''Model''',	font_size=24							)
       UpperCamelCase__							:       Tuple      =   Group(__magic_name__,	__magic_name__							).arrange(__magic_name__,	buff=0.5,	aligned_edge=__magic_name__							)
       model.move_to([3, -1.0, 0]							)
       self.add(__magic_name__							)

       UpperCamelCase__							:       Union[str, Any]      =   []
       UpperCamelCase__							:       Any      =   []
       UpperCamelCase__							:       List[Any]      =   []

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

           UpperCamelCase__							:       Tuple      =   Rectangle(height=0.46 / 4,	width=0.46 / 3							).set_stroke(width=0.0							).set_fill(__magic_name__,	opacity=0.7							)

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

       self.add(*__magic_name__,	*__magic_name__,	*__magic_name__							)

       UpperCamelCase__							:       Tuple      =   [mem.copy() for i in range(6							)]
       UpperCamelCase__							:       Dict      =   VGroup(*__magic_name__							).arrange(__magic_name__,	buff=0							)

       UpperCamelCase__							:       List[str]      =   Text('''Loaded Checkpoint''',	font_size=24							)
       UpperCamelCase__							:       Optional[Any]      =   Group(__magic_name__,	__magic_name__							).arrange(__magic_name__,	buff=0.5,	aligned_edge=__magic_name__							)
       checkpoint.move_to([3, 0.5, 0]							)

       self.add(__magic_name__							)

       UpperCamelCase__							:       Tuple      =   []
       UpperCamelCase__							:       List[str]      =   []

       for i, rect in enumerate(__magic_name__							):
           UpperCamelCase__							:       Optional[int]      =   fill.copy().set_fill(__magic_name__,	opacity=0.7							)
           target.move_to(__magic_name__							)
           ckpt_arr.append(__magic_name__							)

           UpperCamelCase__							:       int      =   target.copy()
           if i < 5:
               cpu_target.move_to(cpu_left_col_base[i + 1]							)
           else:
               cpu_target.move_to(cpu_right_col_base[i - 5]							)
           ckpt_cpu_arr.append(__magic_name__							)
       self.add(*__magic_name__,	*__magic_name__							)

       UpperCamelCase__							:       List[str]      =   Square(side_length=2.2							)
       key.move_to([-5, 2, 0]							)

       UpperCamelCase__							:       List[str]      =   MarkupText(
           f"<b>Key:</b>\n\n<span fgcolor='{YELLOW}'>●</span> Empty Model",	font_size=18,	)

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

       self.add(__magic_name__,	__magic_name__							)

       UpperCamelCase__							:       Any      =   MarkupText(
           f"<span fgcolor='{BLUE}'>●</span> Checkpoint",	font_size=18,	)

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

       UpperCamelCase__							:       Dict      =   MarkupText(
           f"Based on the passed in configuration, weights are stored in\na variety of np.memmaps on disk or to a particular device.",	font_size=24,	)
       step_a.move_to([2, 2, 0]							)

       UpperCamelCase__							:       Any      =   [meta_mem.copy() for i in range(6							)]
       UpperCamelCase__							:       int      =   [meta_mem.copy() for i in range(6							)]
       UpperCamelCase__							:       Optional[int]      =   VGroup(*__magic_name__							).arrange(__magic_name__,	buff=0							)
       UpperCamelCase__							:       Union[str, Any]      =   VGroup(*__magic_name__							).arrange(__magic_name__,	buff=0							)
       UpperCamelCase__							:       Union[str, Any]      =   VGroup(__magic_name__,	__magic_name__							).arrange(__magic_name__,	buff=0							)
       UpperCamelCase__							:       Any      =   Text('''Disk''',	font_size=24							)
       UpperCamelCase__							:       List[Any]      =   Group(__magic_name__,	__magic_name__							).arrange(__magic_name__,	buff=0.5,	aligned_edge=__magic_name__							)
       disk.move_to([-4.0, -1.25, 0]							)
       self.play(Write(__magic_name__,	run_time=3							),	Write(__magic_name__,	run_time=1							),	Create(__magic_name__,	run_time=1							)							)

       UpperCamelCase__							:       Union[str, Any]      =   []
       for i, rect in enumerate(__magic_name__							):
           UpperCamelCase__							:       Union[str, Any]      =   rect.copy()
           target.generate_target()
           target.target.move_to(disk_left_col_base[i]							).scale(0.5							)
           animations.append(MoveToTarget(__magic_name__,	run_time=1.5							)							)
       self.play(*__magic_name__							)

       self.play(FadeOut(__magic_name__							)							)

       UpperCamelCase__							:       Optional[int]      =   MarkupText(f"Then, the checkpoint is removed from memory\nthrough garbage collection.",	font_size=24							)
       step_a.move_to([2, 2, 0]							)

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

       self.play(
           FadeOut(__magic_name__,	__magic_name__,	*__magic_name__,	*__magic_name__							),	)

       self.wait()

