



'''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 get_down_block, get_mid_block, get_out_block, get_up_block



@dataclass
class 							lowercase       (  __a				):


 """simple docstring"""


 UpperCAmelCase							=  42







class 							lowercase       (  __a					,					__a				):


 """simple docstring"""


 @register_to_config
 def __init__(							self  ,a_ = 65_536  ,a_ = None  ,a_ = 2  ,a_ = 2  ,a_ = 0  ,a_ = "fourier"  ,a_ = True  ,a_ = False  ,a_ = 0.0  ,a_ = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")  ,a_ = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")  ,a_ = "UNetMidBlock1D"  ,a_ = None  ,a_ = (32, 32, 64)  ,a_ = None  ,a_ = 8  ,a_ = 1  ,a_ = False  ,)		->   List[str]:
        super().__init__()
        _UpperCAmelCase  :						Optional[int]			     = sample_size

        # time
        if time_embedding_type == "fourier":
               _UpperCAmelCase  :						Tuple			     = GaussianFourierProjection(
                   embedding_size=8  ,set_W_to_weight=__a  ,log=__a  ,flip_sin_to_cos=__a							)
               _UpperCAmelCase  :						str			     = 2 * block_out_channels[0]
        elif time_embedding_type == "positional":
               _UpperCAmelCase  :						Dict			     = Timesteps(
                   block_out_channels[0]  ,flip_sin_to_cos=__a  ,downscale_freq_shift=__a							)
               _UpperCAmelCase  :						Dict			     = block_out_channels[0]

        if use_timestep_embedding:
               _UpperCAmelCase  :						List[Any]			     = block_out_channels[0] * 4
               _UpperCAmelCase  :						Dict			     = TimestepEmbedding(
                   in_channels=__a  ,time_embed_dim=__a  ,act_fn=__a  ,out_dim=block_out_channels[0]  ,)

        _UpperCAmelCase  :						Any			     = nn.ModuleList([]							)
        _UpperCAmelCase  :						Union[str, Any]			     = None
        _UpperCAmelCase  :						str			     = nn.ModuleList([]							)
        _UpperCAmelCase  :						Dict			     = None

        # down
        _UpperCAmelCase  :						Dict			     = in_channels
        for i, down_block_type in enumerate(__a							):
               _UpperCAmelCase  :						Dict			     = output_channel
               _UpperCAmelCase  :						List[Any]			     = block_out_channels[i]

               if i == 0:
                      input_channel += extra_in_channels

               _UpperCAmelCase  :						Any			     = i == len(__a							) - 1

               _UpperCAmelCase  :						Optional[Any]			     = get_down_block(
                   __a  ,num_layers=__a  ,in_channels=__a  ,out_channels=__a  ,temb_channels=block_out_channels[0]  ,add_downsample=not is_final_block or downsample_each_block  ,)
               self.down_blocks.append(__a							)

        # mid
        _UpperCAmelCase  :						Tuple			     = get_mid_block(
            __a  ,in_channels=block_out_channels[-1]  ,mid_channels=block_out_channels[-1]  ,out_channels=block_out_channels[-1]  ,embed_dim=block_out_channels[0]  ,num_layers=__a  ,add_downsample=__a  ,)

        # up
        _UpperCAmelCase  :						List[str]			     = list(reversed(__a							)							)
        _UpperCAmelCase  :						Dict			     = reversed_block_out_channels[0]
        if out_block_type is None:
               _UpperCAmelCase  :						List[Any]			     = out_channels
        else:
               _UpperCAmelCase  :						str			     = block_out_channels[0]

        for i, up_block_type in enumerate(__a							):
               _UpperCAmelCase  :						str			     = output_channel
               _UpperCAmelCase  :						str			     = (
                   reversed_block_out_channels[i + 1] if i < len(__a							) - 1 else final_upsample_channels
               )

               _UpperCAmelCase  :						List[str]			     = i == len(__a							) - 1

               _UpperCAmelCase  :						Any			     = get_up_block(
                   __a  ,num_layers=__a  ,in_channels=__a  ,out_channels=__a  ,temb_channels=block_out_channels[0]  ,add_upsample=not is_final_block  ,)
               self.up_blocks.append(__a							)
               _UpperCAmelCase  :						Optional[int]			     = output_channel

        # out
        _UpperCAmelCase  :						Union[str, Any]			     = norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4  ,32							)
        _UpperCAmelCase  :						int			     = get_out_block(
            out_block_type=__a  ,num_groups_out=__a  ,embed_dim=block_out_channels[0]  ,out_channels=__a  ,act_fn=__a  ,fc_dim=block_out_channels[-1] // 4  ,)







 def _snake_case (							self  ,a_  ,a_  ,a_ = True  ,)		->   Union[UNetaDOutput, Tuple]:
        _UpperCAmelCase  :						Union[str, Any]			     = timestep
        if not torch.is_tensor(__a							):
               _UpperCAmelCase  :						Union[str, Any]			     = torch.tensor([timesteps]  ,dtype=torch.long  ,device=sample.device							)
        elif torch.is_tensor(__a							) and len(timesteps.shape							) == 0:
               _UpperCAmelCase  :						Dict			     = timesteps[None].to(sample.device							)

        _UpperCAmelCase  :						int			     = self.time_proj(__a							)
        if self.config.use_timestep_embedding:
               _UpperCAmelCase  :						List[str]			     = self.time_mlp(__a							)
        else:
               _UpperCAmelCase  :						Optional[Any]			     = timestep_embed[..., None]
               _UpperCAmelCase  :						Tuple			     = timestep_embed.repeat([1, 1, sample.shape[2]]							).to(sample.dtype							)
               _UpperCAmelCase  :						Dict			     = timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:])							)

        # 2. down
        _UpperCAmelCase  :						Tuple			     = ()
        for downsample_block in self.down_blocks:
               _UpperCAmelCase  :						Dict			     = downsample_block(hidden_states=__a  ,temb=__a							)
               down_block_res_samples += res_samples

        # 3. mid
        if self.mid_block:
               _UpperCAmelCase  :						List[Any]			     = self.mid_block(__a  ,__a							)

        # 4. up
        for i, upsample_block in enumerate(self.up_blocks							):
               _UpperCAmelCase  :						Optional[Any]			     = down_block_res_samples[-1:]
               _UpperCAmelCase  :						Tuple			     = down_block_res_samples[:-1]
               _UpperCAmelCase  :						Tuple			     = upsample_block(__a  ,res_hidden_states_tuple=__a  ,temb=__a							)

        # 5. post-process
        if self.out_block:
               _UpperCAmelCase  :						Tuple			     = self.out_block(__a  ,__a							)

        if not return_dict:
               return (sample,)

        return UNetaDOutput(sample=__a							)











# flake8: noqa
# Lint as: python3
lowerCamelCase       :       Optional[Any]        =				[
    '''VerificationMode''',
    '''Version''',
    '''disable_progress_bar''',
    '''enable_progress_bar''',
    '''is_progress_bar_enabled''',
    '''experimental''',
]

from .info_utils import VerificationMode
from .logging import disable_progress_bar, enable_progress_bar, is_progress_bar_enabled
from .version import Version
from .experimental import experimental