






'''simple docstring'''







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      __SCREAMING_SNAKE_CASE					(snake_case_					):


  """simple docstring"""


  _a     :   List[str]			 =    42





class      __SCREAMING_SNAKE_CASE					(snake_case_       ,					snake_case_					):


  """simple docstring"""


  @register_to_config
  def __init__( self      ,  UpperCamelCase__ = None      ,  UpperCamelCase__ = 3      ,  UpperCamelCase__ = 3      ,  UpperCamelCase__ = False      ,  UpperCamelCase__ = "positional"      ,  UpperCamelCase__ = 0      ,  UpperCamelCase__ = True      ,  UpperCamelCase__ = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")      ,  UpperCamelCase__ = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")      ,  UpperCamelCase__ = (224, 448, 672, 896)      ,  UpperCamelCase__ = 2      ,  UpperCamelCase__ = 1      ,  UpperCamelCase__ = 1      ,  UpperCamelCase__ = "conv"      ,  UpperCamelCase__ = "conv"      ,  UpperCamelCase__ = "silu"      ,  UpperCamelCase__ = 8      ,  UpperCamelCase__ = 32      ,  UpperCamelCase__ = 1e-5      ,  UpperCamelCase__ = "default"      ,  UpperCamelCase__ = True      ,  UpperCamelCase__ = None      ,  UpperCamelCase__ = None      ,  ):






         """simple docstring"""


         super().__init__()

         a_	    = sample_size
         a_	    = block_out_channels[0] * 4

         # Check inputs
         if len(UpperCamelCase__  ) != len(UpperCamelCase__  ):
                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(UpperCamelCase__  ) != len(UpperCamelCase__  ):
                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
         a_	    = nn.Convad(UpperCamelCase__      ,  block_out_channels[0]      ,  kernel_size=3      ,  padding=(1, 1)  )

         # time
         if time_embedding_type == "fourier":
                a_	    = GaussianFourierProjection(embedding_size=block_out_channels[0]      ,  scale=16  )
                a_	    = 2 * block_out_channels[0]
         elif time_embedding_type == "positional":
                a_	    = Timesteps(block_out_channels[0]      ,  UpperCamelCase__      ,  UpperCamelCase__  )
                a_	    = block_out_channels[0]

         a_	    = TimestepEmbedding(UpperCamelCase__      ,  UpperCamelCase__  )

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

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

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

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

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

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

                a_	    = i == len(UpperCamelCase__  ) - 1

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

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







  def       _a					( self      ,  UpperCamelCase__      ,  UpperCamelCase__      ,  UpperCamelCase__ = None      ,  UpperCamelCase__ = True      ,  ):






         """simple docstring"""


         if self.config.center_input_sample:
                a_	    = 2 * sample - 1.0

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

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

         a_	    = self.time_proj(UpperCamelCase__  )

         # 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.
         a_	    = t_emb.to(dtype=self.dtype  )
         a_	    = self.time_embedding(UpperCamelCase__  )

         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":
                       a_	    = self.time_proj(UpperCamelCase__  )

                a_	    = self.class_embedding(UpperCamelCase__  ).to(dtype=self.dtype  )
                a_	    = emb + class_emb

         # 2. pre-process
         a_	    = sample
         a_	    = self.conv_in(UpperCamelCase__  )

         # 3. down
         a_	    = (sample,)
         for downsample_block in self.down_blocks:
                if hasattr(UpperCamelCase__      ,  'skip_conv'  ):
                       a_    ,  a_    ,  a_	    = downsample_block(
                           hidden_states=UpperCamelCase__      ,  temb=UpperCamelCase__      ,  skip_sample=UpperCamelCase__  )
                else:
                       a_    ,  a_	    = downsample_block(hidden_states=UpperCamelCase__      ,  temb=UpperCamelCase__  )

                down_block_res_samples += res_samples

         # 4. mid
         a_	    = self.mid_block(UpperCamelCase__      ,  UpperCamelCase__  )

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

                if hasattr(UpperCamelCase__      ,  'skip_conv'  ):
                       a_    ,  a_	    = upsample_block(UpperCamelCase__      ,  UpperCamelCase__      ,  UpperCamelCase__      ,  UpperCamelCase__  )
                else:
                       a_	    = upsample_block(UpperCamelCase__      ,  UpperCamelCase__      ,  UpperCamelCase__  )

        # 6. post-process
         a_	    = self.conv_norm_out(UpperCamelCase__  )
         a_	    = self.conv_act(UpperCamelCase__  )
         a_	    = self.conv_out(UpperCamelCase__  )

         if skip_sample is not None:
                sample += skip_sample

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

         if not return_dict:
                return (sample,)

         return UNetaDOutput(sample=UpperCamelCase__  )












'''simple docstring'''







import re


def 			__UpperCamelCase				(			lowercase_					:     int      ):
       """simple docstring"""




       def docstring_decorator(lowercase_					:     int      ):
              a_	    = fn.__doc__
              a_	    = func_doc.split('\n'      )
              a_	    = 0
              while i < len(lowercase_      ) and re.search(R'^\s*Examples?:\s*$'      ,					lines[i]      ) is None:
                     i += 1
              if i < len(lowercase_      ):
                     a_	    = example_docstring
                     a_	    = '\n'.join(lowercase_      )
              else:
                     raise ValueError(
                         F'The function {fn} should have an empty \'Examples:\' in its docstring as placeholder, '
                         F'current docstring is:\n{func_doc}'      )
              a_	    = func_doc
              return fn

       return docstring_decorator




