



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    ):






						'''simple docstring'''
						a__ =	42







class 						lowerCamelCase_     (			__a    ,__a    ):






						'''simple docstring'''







						@register_to_config
						def __init__(						self   :      List[str]						,						__lowerCamelCase   :      Optional[Union[int, Tuple[int, int]]] = None						,						__lowerCamelCase   :      int = 3						,						__lowerCamelCase   :      int = 3						,						__lowerCamelCase   :      bool = False						,						__lowerCamelCase   :      str = "positional"						,						__lowerCamelCase   :      int = 0						,						__lowerCamelCase   :      bool = True						,						__lowerCamelCase   :      Tuple[str] = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")						,						__lowerCamelCase   :      Tuple[str] = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")						,						__lowerCamelCase   :      Tuple[int] = (2_24, 4_48, 6_72, 8_96)						,						__lowerCamelCase   :      int = 2						,						__lowerCamelCase   :      float = 1						,						__lowerCamelCase   :      int = 1						,						__lowerCamelCase   :      str = "conv"						,						__lowerCamelCase   :      str = "conv"						,						__lowerCamelCase   :      str = "silu"						,						__lowerCamelCase   :      Optional[int] = 8						,						__lowerCamelCase   :      int = 32						,						__lowerCamelCase   :      float = 1e-5						,						__lowerCamelCase   :      str = "default"						,						__lowerCamelCase   :      bool = True						,						__lowerCamelCase   :      Optional[str] = None						,						__lowerCamelCase   :      Optional[int] = None						,						)   ->					List[Any]:
									super().__init__()

									A				:       Any														=    sample_size
									A				:       List[Any]														=    block_out_channels[0] * 4

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

									# time
									if time_embedding_type == "fourier":
												A				:       Optional[Any]														=    GaussianFourierProjection(embedding_size=block_out_channels[0]						,						scale=16			)
												A				:       Optional[Any]														=    2 * block_out_channels[0]
									elif time_embedding_type == "positional":
												A				:       List[Any]														=    Timesteps(block_out_channels[0]						,						__lowerCamelCase						,						__lowerCamelCase			)
												A				:       Dict														=    block_out_channels[0]

									A				:       Optional[Any]														=    TimestepEmbedding(__lowerCamelCase						,						__lowerCamelCase			)

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

									A				:       Dict														=    nn.ModuleList([]			)
									A				:       Any														=    None
									A				:       Dict														=    nn.ModuleList([]			)

									# down
									A				:       Optional[Any]														=    block_out_channels[0]
									for i, down_block_type in enumerate(__lowerCamelCase			):
												A				:       Optional[Any]														=    output_channel
												A				:       Optional[int]														=    block_out_channels[i]
												A				:       Any														=    i == len(__lowerCamelCase			) - 1

												A				:       List[str]														=    get_down_block(
												    __lowerCamelCase						,						num_layers=__lowerCamelCase						,						in_channels=__lowerCamelCase						,						out_channels=__lowerCamelCase						,						temb_channels=__lowerCamelCase						,						add_downsample=not is_final_block						,						resnet_eps=__lowerCamelCase						,						resnet_act_fn=__lowerCamelCase						,						resnet_groups=__lowerCamelCase						,						attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel						,						downsample_padding=__lowerCamelCase						,						resnet_time_scale_shift=__lowerCamelCase						,						downsample_type=__lowerCamelCase						,						)
												self.down_blocks.append(__lowerCamelCase			)

									# mid
									A				:       Optional[int]														=    UNetMidBlockaD(
									    in_channels=block_out_channels[-1]						,						temb_channels=__lowerCamelCase						,						resnet_eps=__lowerCamelCase						,						resnet_act_fn=__lowerCamelCase						,						output_scale_factor=__lowerCamelCase						,						resnet_time_scale_shift=__lowerCamelCase						,						attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1]						,						resnet_groups=__lowerCamelCase						,						add_attention=__lowerCamelCase						,						)

									# up
									A				:       Any														=    list(reversed(__lowerCamelCase			)			)
									A				:       Union[str, Any]														=    reversed_block_out_channels[0]
									for i, up_block_type in enumerate(__lowerCamelCase			):
												A				:       Any														=    output_channel
												A				:       List[str]														=    reversed_block_out_channels[i]
												A				:       Dict														=    reversed_block_out_channels[min(i + 1						,						len(__lowerCamelCase			) - 1			)]

												A				:       Any														=    i == len(__lowerCamelCase			) - 1

												A				:       Optional[int]														=    get_up_block(
												    __lowerCamelCase						,						num_layers=layers_per_block + 1						,						in_channels=__lowerCamelCase						,						out_channels=__lowerCamelCase						,						prev_output_channel=__lowerCamelCase						,						temb_channels=__lowerCamelCase						,						add_upsample=not is_final_block						,						resnet_eps=__lowerCamelCase						,						resnet_act_fn=__lowerCamelCase						,						resnet_groups=__lowerCamelCase						,						attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel						,						resnet_time_scale_shift=__lowerCamelCase						,						upsample_type=__lowerCamelCase						,						)
												self.up_blocks.append(__lowerCamelCase			)
												A				:       Tuple														=    output_channel

									# out
									A				:       Optional[int]														=    norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4						,						32			)
									A				:       List[str]														=    nn.GroupNorm(num_channels=block_out_channels[0]						,						num_groups=__lowerCamelCase						,						eps=__lowerCamelCase			)
									A				:       Tuple														=    nn.SiLU()
									A				:       List[str]														=    nn.Convad(block_out_channels[0]						,						__lowerCamelCase						,						kernel_size=3						,						padding=1			)







						def       SCREAMING_SNAKE_CASE__		(						self   :      Union[str, Any]						,						__lowerCamelCase   :      torch.FloatTensor						,						__lowerCamelCase   :      Union[torch.Tensor, float, int]						,						__lowerCamelCase   :      Optional[torch.Tensor] = None						,						__lowerCamelCase   :      bool = True						,						)   ->					Dict:

									if self.config.center_input_sample:
												A				:       List[Any]														=    2 * sample - 1.0

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

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

									A				:       Any														=    self.time_proj(__lowerCamelCase			)

									# 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				:       Optional[Any]														=    t_emb.to(dtype=self.dtype			)
									A				:       List[Any]														=    self.time_embedding(__lowerCamelCase			)

									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				:       Dict														=    self.time_proj(__lowerCamelCase			)

												A				:       List[Any]														=    self.class_embedding(__lowerCamelCase			).to(dtype=self.dtype			)
												A				:       str														=    emb + class_emb

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

									# 3. down
									A				:       Union[str, Any]														=    (sample,)
									for downsample_block in self.down_blocks:
												if hasattr(__lowerCamelCase						,						"skip_conv"			):
															A							,			A							,			A				:       Union[str, Any]														=    downsample_block(
															    hidden_states=__lowerCamelCase						,						temb=__lowerCamelCase						,						skip_sample=__lowerCamelCase			)
												else:
															A							,			A				:       Tuple														=    downsample_block(hidden_states=__lowerCamelCase						,						temb=__lowerCamelCase			)

												down_block_res_samples += res_samples

									# 4. mid
									A				:       Union[str, Any]														=    self.mid_block(__lowerCamelCase						,						__lowerCamelCase			)

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

												if hasattr(__lowerCamelCase						,						"skip_conv"			):
															A							,			A				:       Dict														=    upsample_block(__lowerCamelCase						,						__lowerCamelCase						,						__lowerCamelCase						,						__lowerCamelCase			)
												else:
															A				:       Any														=    upsample_block(__lowerCamelCase						,						__lowerCamelCase						,						__lowerCamelCase			)

        # 6. post-process
									A				:       Dict														=    self.conv_norm_out(__lowerCamelCase			)
									A				:       Optional[Any]														=    self.conv_act(__lowerCamelCase			)
									A				:       str														=    self.conv_out(__lowerCamelCase			)

									if skip_sample is not None:
												sample += skip_sample

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

									if not return_dict:
												return (sample,)

									return UNetaDOutput(sample=__lowerCamelCase			)





'''simple docstring'''







import json
import os
from dataclasses import dataclass

import numpy as np
import pyarrow as pa

import datasets
from utils import get_duration


A					       =10_00_00_00_00_00
A					       =1_00_00

A       , A					       =os.path.split(__file__)
A					       =os.path.join(RESULTS_BASEPATH, 'results', RESULTS_FILENAME.replace('.py', '.json'))





def 		snake_case_				(_a      :				int       ,  _a      :				int	):
    UpperCAmelCase     			=						pa.Table.from_pydict({'''col''': [0] * chunk_size}	)
    UpperCAmelCase     			=						pa.concat_tables([table] * (num_examples // chunk_size)	)
    return datasets.Dataset(_a       ,  fingerprint='''table_100B'''	)


@dataclass
class      _a    :
       __a							:	int
       __a							:	int
       __a							:	int
       __a							:	int
       def        A (      self		:      List[str]							):



           '''simple docstring'''

           UpperCAmelCase     			=						np.random.default_rng(self.seed							)
           UpperCAmelCase     			=						rng.integers(low=self.low  ,							high=self.high  ,							size=self.size							).tolist()
       def __iter__(      self		:      Tuple							):



           '''simple docstring'''

           return iter(self._sampled_values							)







       def __len__(      self		:      Dict							):



           '''simple docstring'''

           return self.size






@get_duration
def 		snake_case_				(_a      :				datasets.Dataset	):
    UpperCAmelCase     			=						dataset[0]





@get_duration
def 		snake_case_				(_a      :				datasets.Dataset	):
    UpperCAmelCase     			=						dataset[-1]





@get_duration
def 		snake_case_				(_a      :				datasets.Dataset	):
    UpperCAmelCase     			=						dataset[range(len(_a	) // 2       ,  len(_a	) // 2 + 1_0_2_4	)]





@get_duration
def 		snake_case_				(_a      :				datasets.Dataset	):
    UpperCAmelCase     			=						dataset[RandIter(0       ,  len(_a	)       ,  1_0_2_4       ,  seed=4_2	)]





def 		snake_case_				():
    UpperCAmelCase     			=						{'''num examples''': SPEED_TEST_N_EXAMPLES}
    UpperCAmelCase     			=						(get_first_row, get_last_row, get_batch_of_aaaa_rows, get_batch_of_aaaa_random_rows)
    print('''generating dataset'''	)
    UpperCAmelCase     			=						generate_100B_dataset(num_examples=_a       ,  chunk_size=_a	)
    print('''Functions'''	)
    for func in functions:
        print(func.__name__	)
        UpperCAmelCase     			=						func(_a	)

    with open(_a       ,  '''wb'''	) as f:
        f.write(json.dumps(_a	).encode('''utf-8'''	)	)


if __name__ == "__main__":  # useful to run the profiler
   benchmark_table_100B()




