





'''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 					snake_case					(   SCREAMING_SNAKE_CASE__	):




 """simple docstring"""



 SCREAMING_SNAKE_CASE_     : torch.FloatTensor






class 					snake_case					(   SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__	):




 """simple docstring"""



 @register_to_config
 def __init__(	self	:							Tuple							,     __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							,     ):
        super().__init__()

        __UpperCamelCase	=  sample_size
        __UpperCamelCase	=  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
        __UpperCamelCase	=  nn.Convad(__A							,     block_out_channels[0]							,     kernel_size=3							,     padding=(1, 1)     )

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

        __UpperCamelCase	=  TimestepEmbedding(__A							,     __A     )

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

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

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

               __UpperCamelCase	=  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
        __UpperCamelCase	=  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
        __UpperCamelCase	=  list(reversed(__A     )     )
        __UpperCamelCase	=  reversed_block_out_channels[0]
        for i, up_block_type in enumerate(__A     ):
               __UpperCamelCase	=  output_channel
               __UpperCamelCase	=  reversed_block_out_channels[i]
               __UpperCamelCase	=  reversed_block_out_channels[min(i + 1							,     len(__A     ) - 1     )]

               __UpperCamelCase	=  i == len(__A     ) - 1

               __UpperCamelCase	=  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     )
               __UpperCamelCase	=  output_channel

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






 def 	_lowerCamelCase (	self	:							Optional[Any]							,     __A	:							torch.FloatTensor							,     __A	:							Union[torch.Tensor, float, int]							,     __A	:							Optional[torch.Tensor] = None							,     __A	:							bool = True							,     ):

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

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

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

        __UpperCamelCase	=  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.
        __UpperCamelCase	=  t_emb.to(dtype=self.dtype     )
        __UpperCamelCase	=  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":
                      __UpperCamelCase	=  self.time_proj(__A     )

               __UpperCamelCase	=  self.class_embedding(__A     ).to(dtype=self.dtype     )
               __UpperCamelCase	=  emb + class_emb

        # 2. pre-process
        __UpperCamelCase	=  sample
        __UpperCamelCase	=  self.conv_in(__A     )

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

               down_block_res_samples += res_samples

        # 4. mid
        __UpperCamelCase	=  self.mid_block(__A							,     __A     )

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

               if hasattr(__A							,     'skip_conv'     ):
                      __UpperCamelCase	,						__UpperCamelCase	=  upsample_block(__A							,     __A							,     __A							,     __A     )
               else:
                      __UpperCamelCase	=  upsample_block(__A							,     __A							,     __A     )

        # 6. post-process
        __UpperCamelCase	=  self.conv_norm_out(__A     )
        __UpperCamelCase	=  self.conv_act(__A     )
        __UpperCamelCase	=  self.conv_out(__A     )

        if skip_sample is not None:
               sample += skip_sample

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

        if not return_dict:
               return (sample,)

        return UNetaDOutput(sample=__A     )







'''simple docstring'''
import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

from tqdm import auto as tqdm_lib


a__							:							Optional[int]			=threading.Lock()
a__							:							Optional[logging.Handler]			=None

a__							:							Optional[Any]			={
    '''debug''': logging.DEBUG,
    '''info''': logging.INFO,
    '''warning''': logging.WARNING,
    '''error''': logging.ERROR,
    '''critical''': logging.CRITICAL,
}

a__							:							List[Any]			=logging.WARNING

a__							:							str			=True
def     lowercase__       (       )			->				int:



       """simple docstring"""






       __UpperCamelCase	=  os.getenv('DIFFUSERS_VERBOSITY'    ,  __lowercase					)
       if env_level_str:
              if env_level_str in log_levels:
                     return log_levels[env_level_str]
              else:
                     logging.getLogger().warning(
                         F'''Unknown option DIFFUSERS_VERBOSITY={env_level_str}, '''
                         F'''has to be one of: { ', '.join(log_levels.keys()					) }'''					)
       return _default_log_level
def     lowercase__       (       )			->				str:



       """simple docstring"""






       return __name__.split('.'					)[0]
def     lowercase__       (       )			->				logging.Logger:



       """simple docstring"""






       return logging.getLogger(_get_library_name()					)
def     lowercase__       (       )			->				None:



       """simple docstring"""






       global _default_handler

       with _lock:
              if _default_handler:
                     # This library has already configured the library root logger.
                     return
              __UpperCamelCase	=  logging.StreamHandler()  # Set sys.stderr as stream.
              __UpperCamelCase	=  sys.stderr.flush

              # Apply our default configuration to the library root logger.
              __UpperCamelCase	=  _get_library_root_logger()
              library_root_logger.addHandler(_default_handler					)
              library_root_logger.setLevel(_get_default_logging_level()					)
              __UpperCamelCase	=  False
def     lowercase__       (       )			->				None:



       """simple docstring"""






       global _default_handler

       with _lock:
              if not _default_handler:
                     return

              __UpperCamelCase	=  _get_library_root_logger()
              library_root_logger.removeHandler(_default_handler					)
              library_root_logger.setLevel(logging.NOTSET					)
              __UpperCamelCase	=  None
def     lowercase__       (       )			->				str:



       """simple docstring"""






       return log_levels
def     lowercase__       (       __lowercase						:  Optional[str] = None					)			->				logging.Logger:



       """simple docstring"""






       if name is None:
              __UpperCamelCase	=  _get_library_name()

       _configure_library_root_logger()
       return logging.getLogger(__lowercase					)
def     lowercase__       (       )			->				int:



       """simple docstring"""






       _configure_library_root_logger()
       return _get_library_root_logger().getEffectiveLevel()
def     lowercase__       (       __lowercase						:  int					)			->				None:



       """simple docstring"""






       _configure_library_root_logger()
       _get_library_root_logger().setLevel(__lowercase					)
def     lowercase__       (       )			->				Tuple:



       """simple docstring"""






       return set_verbosity(__lowercase					)
def     lowercase__       (       )			->				Union[str, Any]:



       """simple docstring"""






       return set_verbosity(__lowercase					)
def     lowercase__       (       )			->				Any:



       """simple docstring"""






       return set_verbosity(__lowercase					)
def     lowercase__       (       )			->				int:



       """simple docstring"""






       return set_verbosity(__lowercase					)
def     lowercase__       (       )			->				None:



       """simple docstring"""






       _configure_library_root_logger()

       assert _default_handler is not None
       _get_library_root_logger().removeHandler(_default_handler					)
def     lowercase__       (       )			->				None:



       """simple docstring"""






       _configure_library_root_logger()

       assert _default_handler is not None
       _get_library_root_logger().addHandler(_default_handler					)
def     lowercase__       (       __lowercase						:  logging.Handler					)			->				None:



       """simple docstring"""






       _configure_library_root_logger()

       assert handler is not None
       _get_library_root_logger().addHandler(__lowercase					)
def     lowercase__       (       __lowercase						:  logging.Handler					)			->				None:



       """simple docstring"""






       _configure_library_root_logger()

       assert handler is not None and handler not in _get_library_root_logger().handlers
       _get_library_root_logger().removeHandler(__lowercase					)
def     lowercase__       (       )			->				None:



       """simple docstring"""






       _configure_library_root_logger()
       __UpperCamelCase	=  False
def     lowercase__       (       )			->				None:



       """simple docstring"""






       _configure_library_root_logger()
       __UpperCamelCase	=  True
def     lowercase__       (       )			->				None:



       """simple docstring"""






       __UpperCamelCase	=  _get_library_root_logger().handlers

       for handler in handlers:
              __UpperCamelCase	=  logging.Formatter('[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s'					)
              handler.setFormatter(__lowercase					)
def     lowercase__       (       )			->				None:



       """simple docstring"""






       __UpperCamelCase	=  _get_library_root_logger().handlers

       for handler in handlers:
              handler.setFormatter(__lowercase					)
def     lowercase__       (       self						:  int    ,  *__lowercase						:  List[str]    ,  **__lowercase						:  List[Any]					)			->				Any:



       """simple docstring"""






       __UpperCamelCase	=  os.getenv('DIFFUSERS_NO_ADVISORY_WARNINGS'    ,  __lowercase					)
       if no_advisory_warnings:
              return
       self.warning(*__lowercase    ,  **__lowercase					)


a__							:							Dict			=warning_advice

class 					snake_case					:




 """simple docstring"""



 def __init__(	self	:							Optional[int]							,     *__A	:							int							,     **__A	:							Dict     ):  # pylint: disable=unused-argument
        __UpperCamelCase	=  args[0] if args else None






 def __iter__(	self	:							str     ):
        return iter(self._iterator     )






 def __getattr__(	self	:							Tuple							,     __A	:							Any     ):

        def empty_fn(*__A	:							Any							,     **__A	:							List[str]     ):  # pylint: disable=unused-argument
               return

        return empty_fn






 def __enter__(	self	:							str     ):
        return self






 def __exit__(	self	:							Optional[Any]							,     __A	:							Dict							,     __A	:							str							,     __A	:							Dict     ):
        return






class 					snake_case					:




 """simple docstring"""



 def __call__(	self	:							Union[str, Any]							,     *__A	:							Union[str, Any]							,     **__A	:							Dict     ):
        if _tqdm_active:
               return tqdm_lib.tqdm(*__A							,     **__A     )
        else:
               return EmptyTqdm(*__A							,     **__A     )






 def 	_lowerCamelCase (	self	:							Any							,     *__A	:							str							,     **__A	:							Optional[int]     ):
        __UpperCamelCase	=  None
        if _tqdm_active:
               return tqdm_lib.tqdm.set_lock(*__A							,     **__A     )






 def 	_lowerCamelCase (	self	:							Optional[int]     ):
        if _tqdm_active:
               return tqdm_lib.tqdm.get_lock()






a__							:							List[Any]			=_tqdm_cls()
def     lowercase__       (       )			->				bool:



       """simple docstring"""






       global _tqdm_active
       return bool(_tqdm_active					)
def     lowercase__       (       )			->				List[str]:



       """simple docstring"""






       global _tqdm_active
       __UpperCamelCase	=  True


def     lowercase__       (       )			->				str:



       """simple docstring"""






       global _tqdm_active
       __UpperCamelCase	=  False
