






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 							A_      (  __lowercase     ):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:     torch.FloatTensor



class 							A_      (  __lowercase	,				__lowercase     ):






 '''simple docstring'''




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

     """simple docstring"""



     super().__init__()

     _UpperCAmelCase			: Optional[int]     					=			sample_size
     _UpperCAmelCase			: Any     					=			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			: Any     					=			nn.Convad(_A							,					block_out_channels[0]							,					kernel_size=3							,					padding=(1, 1))

     # time
     if time_embedding_type == "fourier":
         _UpperCAmelCase			: str     					=			GaussianFourierProjection(embedding_size=block_out_channels[0]							,					scale=16)
         _UpperCAmelCase			: List[Any]     					=			2 * block_out_channels[0]
     elif time_embedding_type == "positional":
         _UpperCAmelCase			: Any     					=			Timesteps(block_out_channels[0]							,					_A							,					_A)
         _UpperCAmelCase			: int     					=			block_out_channels[0]

     _UpperCAmelCase			: Tuple     					=			TimestepEmbedding(_A							,					_A)

     # class embedding
     if class_embed_type is None and num_class_embeds is not None:
         _UpperCAmelCase			: List[str]     					=			nn.Embedding(_A							,					_A)
     elif class_embed_type == "timestep":
         _UpperCAmelCase			: Optional[int]     					=			TimestepEmbedding(_A							,					_A)
     elif class_embed_type == "identity":
         _UpperCAmelCase			: Any     					=			nn.Identity(_A							,					_A)
     else:
         _UpperCAmelCase			: List[str]     					=			None

     _UpperCAmelCase			: Any     					=			nn.ModuleList([])
     _UpperCAmelCase			: Optional[Any]     					=			None
     _UpperCAmelCase			: Tuple     					=			nn.ModuleList([])

     # down
     _UpperCAmelCase			: List[str]     					=			block_out_channels[0]
     for i, down_block_type in enumerate(_A):
         _UpperCAmelCase			: Optional[Any]     					=			output_channel
         _UpperCAmelCase			: Optional[Any]     					=			block_out_channels[i]
         _UpperCAmelCase			: Any     					=			i == len(_A) - 1

         _UpperCAmelCase			: Optional[int]     					=			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			: Optional[int]     					=			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			: Union[str, Any]     					=			list(reversed(_A))
     _UpperCAmelCase			: Any     					=			reversed_block_out_channels[0]
     for i, up_block_type in enumerate(_A):
         _UpperCAmelCase			: Union[str, Any]     					=			output_channel
         _UpperCAmelCase			: Optional[Any]     					=			reversed_block_out_channels[i]
         _UpperCAmelCase			: Tuple     					=			reversed_block_out_channels[min(i + 1							,					len(_A) - 1)]

         _UpperCAmelCase			: List[Any]     					=			i == len(_A) - 1

         _UpperCAmelCase			: List[str]     					=			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			: Union[str, Any]     					=			output_channel

     # out
     _UpperCAmelCase			: Optional[int]     					=			norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4							,					32)
     _UpperCAmelCase			: int     					=			nn.GroupNorm(num_channels=block_out_channels[0]							,					num_groups=_A							,					eps=_A)
     _UpperCAmelCase			: Dict     					=			nn.SiLU()
     _UpperCAmelCase			: Optional[Any]     					=			nn.Convad(block_out_channels[0]							,					_A							,					kernel_size=3							,					padding=1)


 def 	snake_case__			(		self							,					_A							,					_A							,					_A = None							,					_A = True							,					)	->						Union[UNetaDOutput, Tuple]:

     """simple docstring"""



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

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

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

     _UpperCAmelCase			: Dict     					=			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			: Optional[int]     					=			t_emb.to(dtype=self.dtype)
     _UpperCAmelCase			: Dict     					=			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			: List[str]     					=			self.time_proj(_A)

         _UpperCAmelCase			: Optional[Any]     					=			self.class_embedding(_A).to(dtype=self.dtype)
         _UpperCAmelCase			: Dict     					=			emb + class_emb

     # 2. pre-process
     _UpperCAmelCase			: Tuple     					=			sample
     _UpperCAmelCase			: List[Any]     					=			self.conv_in(_A)

     # 3. down
     _UpperCAmelCase			: Union[str, Any]     					=			(sample,)
     for downsample_block in self.down_blocks:
         if hasattr(_A							,					'''skip_conv'''):
             _UpperCAmelCase					,	_UpperCAmelCase					,	_UpperCAmelCase			: int     					=			downsample_block(
                 hidden_states=_A							,					temb=_A							,					skip_sample=_A)
         else:
             _UpperCAmelCase					,	_UpperCAmelCase			: List[str]     					=			downsample_block(hidden_states=_A							,					temb=_A)

         down_block_res_samples += res_samples

     # 4. mid
     _UpperCAmelCase			: Any     					=			self.mid_block(_A							,					_A)

     # 5. up
     _UpperCAmelCase			: Tuple     					=			None
     for upsample_block in self.up_blocks:
         _UpperCAmelCase			: List[str]     					=			down_block_res_samples[-len(upsample_block.resnets) :]
         _UpperCAmelCase			: Any     					=			down_block_res_samples[: -len(upsample_block.resnets)]

         if hasattr(_A							,					'''skip_conv'''):
             _UpperCAmelCase					,	_UpperCAmelCase			: Union[str, Any]     					=			upsample_block(_A							,					_A							,					_A							,					_A)
         else:
             _UpperCAmelCase			: List[str]     					=			upsample_block(_A							,					_A							,					_A)

        # 6. post-process
     _UpperCAmelCase			: List[str]     					=			self.conv_norm_out(_A)
     _UpperCAmelCase			: Union[str, Any]     					=			self.conv_act(_A)
     _UpperCAmelCase			: str     					=			self.conv_out(_A)

     if skip_sample is not None:
         sample += skip_sample

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

     if not return_dict:
         return (sample,)

     return UNetaDOutput(sample=_A)












from unittest.mock import patch

import datasets
from datasets import Dataset

def        _lowerCamelCase		(  ) ->   Tuple:
    _UpperCAmelCase			: Union[str, Any]     					=			Dataset.from_dict({'''col_1''': [3, 2, 0, 1]}	)

    with patch('''tqdm.auto.tqdm'''	) as mock_tqdm:
        datasets.disable_progress_bar()
        dset.map(lambda __A	: {"col_2": x["col_1"] + 1}	)
        mock_tqdm.assert_not_called()

        mock_tqdm.reset_mock()

        datasets.enable_progress_bar()
        dset.map(lambda __A	: {"col_2": x["col_1"] + 1}	)
        mock_tqdm.assert_called()




