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

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block







@dataclass
class 	lowerCamelCase						(    A_    ):
     UpperCAmelCase__							:				torch.FloatTensor







class 	lowerCamelCase						(    A_				,					A_    ):





     @register_to_config
     def __init__(self :       Optional[int]							,       _A :       Optional[Union[int, Tuple[int, int]]] = None							,       _A :       int = 3							,       _A :       int = 3							,       _A :       bool = False							,       _A :       str = "positional"							,       _A :       int = 0							,       _A :       bool = True							,       _A :       Tuple[str] = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")							,       _A :       Tuple[str] = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")							,       _A :       Tuple[int] = (2_2_4, 4_4_8, 6_7_2, 8_9_6)							,       _A :       int = 2							,       _A :       float = 1							,       _A :       int = 1							,       _A :       str = "conv"							,       _A :       str = "conv"							,       _A :       str = "silu"							,       _A :       Optional[int] = 8							,       _A :       int = 3_2							,       _A :       float = 1E-5							,       _A :       str = "default"							,       _A :       bool = True							,       _A :       Optional[str] = None							,       _A :       Optional[int] = None							,       )		->		Any:
         super().__init__()

         snake_case						=   sample_size
         snake_case						=   block_out_channels[0] * 4

         # Check inputs
         if len(_A  ) != len(_A  ):
             raise ValueError(
                 f'Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}.'  )

         if len(_A  ) != len(_A  ):
             raise ValueError(
                 f'Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}.'  )

         # input
         snake_case						=   nn.Convad(_A							,       block_out_channels[0]							,       kernel_size=3							,       padding=(1, 1)  )

         # time
         if time_embedding_type == "fourier":
             snake_case						=   GaussianFourierProjection(embedding_size=block_out_channels[0]							,       scale=1_6  )
             snake_case						=   2 * block_out_channels[0]
         elif time_embedding_type == "positional":
             snake_case						=   Timesteps(block_out_channels[0]							,       _A							,       _A  )
             snake_case						=   block_out_channels[0]

         snake_case						=   TimestepEmbedding(_A							,       _A  )

         # class embedding
         if class_embed_type is None and num_class_embeds is not None:
             snake_case						=   nn.Embedding(_A							,       _A  )
         elif class_embed_type == "timestep":
             snake_case						=   TimestepEmbedding(_A							,       _A  )
         elif class_embed_type == "identity":
             snake_case						=   nn.Identity(_A							,       _A  )
         else:
             snake_case						=   None

         snake_case						=   nn.ModuleList([]  )
         snake_case						=   None
         snake_case						=   nn.ModuleList([]  )

         # down
         snake_case						=   block_out_channels[0]
         for i, down_block_type in enumerate(_A  ):
             snake_case						=   output_channel
             snake_case						=   block_out_channels[i]
             snake_case						=   i == len(_A  ) - 1

             snake_case						=   get_down_block(
                 _A							,       num_layers=_A							,       in_channels=_A							,       out_channels=_A							,       temb_channels=_A							,       add_downsample=not is_final_block							,       resnet_eps=_A							,       resnet_act_fn=_A							,       resnet_groups=_A							,       attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel							,       downsample_padding=_A							,       resnet_time_scale_shift=_A							,       downsample_type=_A							,       )
             self.down_blocks.append(_A  )

         # mid
         snake_case						=   UNetMidBlockaD(
             in_channels=block_out_channels[-1]							,       temb_channels=_A							,       resnet_eps=_A							,       resnet_act_fn=_A							,       output_scale_factor=_A							,       resnet_time_scale_shift=_A							,       attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1]							,       resnet_groups=_A							,       add_attention=_A							,       )

         # up
         snake_case						=   list(reversed(_A  )  )
         snake_case						=   reversed_block_out_channels[0]
         for i, up_block_type in enumerate(_A  ):
             snake_case						=   output_channel
             snake_case						=   reversed_block_out_channels[i]
             snake_case						=   reversed_block_out_channels[min(i + 1							,       len(_A  ) - 1  )]

             snake_case						=   i == len(_A  ) - 1

             snake_case						=   get_up_block(
                 _A							,       num_layers=layers_per_block + 1							,       in_channels=_A							,       out_channels=_A							,       prev_output_channel=_A							,       temb_channels=_A							,       add_upsample=not is_final_block							,       resnet_eps=_A							,       resnet_act_fn=_A							,       resnet_groups=_A							,       attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel							,       resnet_time_scale_shift=_A							,       upsample_type=_A							,       )
             self.up_blocks.append(_A  )
             snake_case						=   output_channel

         # out
         snake_case						=   norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4							,       3_2  )
         snake_case						=   nn.GroupNorm(num_channels=block_out_channels[0]							,       num_groups=_A							,       eps=_A  )
         snake_case						=   nn.SiLU()
         snake_case						=   nn.Convad(block_out_channels[0]							,       _A							,       kernel_size=3							,       padding=1  )





     def 						UpperCAmelCase(self :       Dict							,       _A :       torch.FloatTensor							,       _A :       Union[torch.Tensor, float, int]							,       _A :       Optional[torch.Tensor] = None							,       _A :       bool = True							,       )		->		Union[UNetaDOutput, Tuple]:
         # 0. center input if necessary
         if self.config.center_input_sample:
             snake_case						=   2 * sample - 1.0

         # 1. time
         snake_case						=   timestep
         if not torch.is_tensor(_A  ):
             snake_case						=   torch.tensor([timesteps]							,       dtype=torch.long							,       device=sample.device  )
         elif torch.is_tensor(_A  ) and len(timesteps.shape  ) == 0:
             snake_case						=   timesteps[None].to(sample.device  )

         # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
         snake_case						=   timesteps * torch.ones(sample.shape[0]							,       dtype=timesteps.dtype							,       device=timesteps.device  )

         snake_case						=   self.time_proj(_A  )

         # timesteps does not contain any weights and will always return f32 tensors
         # but time_embedding might actually be running in fp16. so we need to cast here.
         # there might be better ways to encapsulate this.
         snake_case						=   t_emb.to(dtype=self.dtype  )
         snake_case						=   self.time_embedding(_A  )

         if self.class_embedding is not None:
             if class_labels is None:
                 raise ValueError("class_labels should be provided when doing class conditioning"  )

             if self.config.class_embed_type == "timestep":
                 snake_case						=   self.time_proj(_A  )

             snake_case						=   self.class_embedding(_A  ).to(dtype=self.dtype  )
             snake_case						=   emb + class_emb

         # 2. pre-process
         snake_case						=   sample
         snake_case						=   self.conv_in(_A  )

         # 3. down
         snake_case						=   (sample,)
         for downsample_block in self.down_blocks:
             if hasattr(_A							,       "skip_conv"  ):
                 snake_case	,				snake_case	,				snake_case						=   downsample_block(
                     hidden_states=_A							,       temb=_A							,       skip_sample=_A  )
             else:
                 snake_case	,				snake_case						=   downsample_block(hidden_states=_A							,       temb=_A  )

             down_block_res_samples += res_samples

         # 4. mid
         snake_case						=   self.mid_block(_A							,       _A  )

         # 5. up
         snake_case						=   None
         for upsample_block in self.up_blocks:
             snake_case						=   down_block_res_samples[-len(upsample_block.resnets  ) :]
             snake_case						=   down_block_res_samples[: -len(upsample_block.resnets  )]

             if hasattr(_A							,       "skip_conv"  ):
                 snake_case	,				snake_case						=   upsample_block(_A							,       _A							,       _A							,       _A  )
             else:
                 snake_case						=   upsample_block(_A							,       _A							,       _A  )

        # 6. post-process
         snake_case						=   self.conv_norm_out(_A  )
         snake_case						=   self.conv_act(_A  )
         snake_case						=   self.conv_out(_A  )

         if skip_sample is not None:
             sample += skip_sample

         if self.config.time_embedding_type == "fourier":
             snake_case						=   timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:]  )))  )
             snake_case						=   sample / timesteps

         if not return_dict:
             return (sample,)

         return UNetaDOutput(sample=_A  )



from ...configuration_utils import PretrainedConfig
from ...utils import logging


_A       =					logging.get_logger(__name__)

_A       =					{
    "openbmb/cpm-ant-10b": "https://huggingface.co/openbmb/cpm-ant-10b/blob/main/config.json"
    # See all CPMAnt models at https://huggingface.co/models?filter=cpmant
}







class 	lowerCamelCase						(    A_    ):
     UpperCAmelCase__							:				List[Any] 	=						"cpmant"





     def __init__(self :       List[Any]							,       _A :       int = 3_0_7_2_0							,       _A :       int = 4_0_9_6							,       _A :       int = 3_2							,       _A :       int = 1_2_8							,       _A :       int = 1_0_2_4_0							,       _A :       int = 4_8							,       _A :       int = 0.0							,       _A :       int = 5_1_2							,       _A :       int = 2_0_4_8							,       _A :       int = 1E-6							,       _A :       float = 1.0							,       _A :       int = 3_2							,       _A :       int = 3_2							,       _A :       int = 3_2							,       _A :       bool = True							,       **_A :       Dict							,       )		->		Dict:
         super().__init__(**_A  )
         snake_case						=   prompt_types
         snake_case						=   prompt_length
         snake_case						=   segment_types
         snake_case						=   hidden_size
         snake_case						=   num_attention_heads
         snake_case						=   dim_head
         snake_case						=   dim_ff
         snake_case						=   num_hidden_layers
         snake_case						=   position_bias_num_buckets
         snake_case						=   position_bias_max_distance
         snake_case						=   dropout_p
         snake_case						=   eps
         snake_case						=   use_cache
         snake_case						=   vocab_size
         snake_case						=   init_std

