





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 	_UpperCAmelCase						( __UpperCAmelCase       ):
					__A        = 42




class 	_UpperCAmelCase						( __UpperCAmelCase			,							__UpperCAmelCase       ):




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

							UpperCamelCase__								=						sample_size
							UpperCamelCase__								=						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
							UpperCamelCase__								=						nn.Convad(UpperCAmelCase_     ,   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=16							)
									UpperCamelCase__								=						2 * block_out_channels[0]
							elif time_embedding_type == "positional":
									UpperCamelCase__								=						Timesteps(block_out_channels[0]     ,   UpperCAmelCase_     ,   UpperCAmelCase_							)
									UpperCamelCase__								=						block_out_channels[0]

							UpperCamelCase__								=						TimestepEmbedding(UpperCAmelCase_     ,   UpperCAmelCase_							)

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

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

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

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

									UpperCamelCase__								=						i == len(UpperCAmelCase_							) - 1

									UpperCamelCase__								=						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_							)
									UpperCamelCase__								=						output_channel

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




					def 						snake_case_		(    self     ,   __lowerCAmelCase     ,   __lowerCAmelCase     ,   __lowerCAmelCase = None     ,   __lowerCAmelCase = True     ,   )    ->	str:
							# 0. center input if necessary
							if self.config.center_input_sample:
									UpperCamelCase__								=						2 * sample - 1.0

							# 1. time
							UpperCamelCase__								=						timestep
							if not torch.is_tensor(UpperCAmelCase_							):
									UpperCamelCase__								=						torch.tensor([timesteps]     ,   dtype=torch.long     ,   device=sample.device							)
							elif torch.is_tensor(UpperCAmelCase_							) 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(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.
							UpperCamelCase__								=						t_emb.to(dtype=self.dtype							)
							UpperCamelCase__								=						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":
											UpperCamelCase__								=						self.time_proj(UpperCAmelCase_							)

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

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

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

									down_block_res_samples += res_samples

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

							# 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(UpperCAmelCase_     ,   """skip_conv"""							):
											UpperCamelCase__	,      UpperCamelCase__								=						upsample_block(UpperCAmelCase_     ,   UpperCAmelCase_     ,   UpperCAmelCase_     ,   UpperCAmelCase_							)
									else:
											UpperCamelCase__								=						upsample_block(UpperCAmelCase_     ,   UpperCAmelCase_     ,   UpperCAmelCase_							)

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

							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=UpperCAmelCase_							)











import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
		import torch


UpperCAmelCase_    =logging.get_logger(__name__)


class 				__UpperCamelCase      :





				'''simple docstring'''



				def __init__(			self							,			UpperCAmelCase_							,			UpperCAmelCase_=21							,			UpperCAmelCase_=[]							,			UpperCAmelCase_=None    ):
											lowerCAmelCase =		model
											lowerCAmelCase =		trace_batch_nums
											lowerCAmelCase =		abort_after_batch_num

											# keep a LIFO buffer of frames to dump as soon as inf/nan is encountered to give context to the problem emergence
											lowerCAmelCase =		collections.deque([]							,			UpperCAmelCase_    )
											lowerCAmelCase =		[]
											lowerCAmelCase =		0
											lowerCAmelCase =		0
											lowerCAmelCase =		False
											lowerCAmelCase =		'''                 '''

											self.analyse_model()

											self.register_forward_hook()



				def    __snake_case		(			self							,			UpperCAmelCase_=None    ):
											if frame is not None:
																		self.expand_frame(UpperCAmelCase_    )
											self.frames.append('''\n'''.join(self.frame    )    )
											lowerCAmelCase =		[]  # start a new frame



				def    __snake_case		(			self							,			UpperCAmelCase_    ):
											self.frame.append(UpperCAmelCase_    )



				def    __snake_case		(			self    ):
											print('''\n'''.join(self.frames    )    )
											lowerCAmelCase =		[]



				def    __snake_case		(			self    ):
											lowerCAmelCase =		[]



				def    __snake_case		(			self    ):
											print(F"""\nDetected inf/nan during batch_number={self.batch_number}"""    )
											print(F"""Last {len(self.frames    )} forward frames:"""    )
											print(F"""{"abs min":8} {"abs max":8} metadata"""    )
											print('''\n'''.join(self.frames    )    )
											print('''\n\n'''    )
											lowerCAmelCase =		[]



				def    __snake_case		(			self    ):
											# extract the fully qualified module names, to be able to report at run time. e.g.:
											# encoder.block.2.layer.0.SelfAttention.o
											#
											# for shared weights only the first shared module name will be registered
											lowerCAmelCase =		{m: name for name, m in self.model.named_modules()}
											# self.longest_module_name = max(len(v) for v in self.module_names.values())



				def    __snake_case		(			self							,			UpperCAmelCase_							,			UpperCAmelCase_    ):
											if torch.is_tensor(UpperCAmelCase_    ):
																		self.expand_frame(get_abs_min_max(UpperCAmelCase_							,			UpperCAmelCase_    )    )
																		if detect_overflow(UpperCAmelCase_							,			UpperCAmelCase_    ):
																									lowerCAmelCase =		True
											elif var is None:
																		self.expand_frame(F"""{"None":>17} {ctx}"""    )
											else:
																		self.expand_frame(F"""{"not a tensor":>17} {ctx}"""    )



				def    __snake_case		(			self    ):
											self.expand_frame(F"""\n\n{self.prefix} *** Starting batch number={self.batch_number} ***"""    )
											self.expand_frame(F"""{"abs min":8} {"abs max":8} metadata"""    )



				def    __snake_case		(			self    ):
											self.expand_frame(F"""{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"""    )



				def    __snake_case		(			self							,			UpperCAmelCase_							,			UpperCAmelCase_							,			UpperCAmelCase_    ):
											self.expand_frame(F"""{self.prefix} {self.module_names[module]} {module.__class__.__name__}"""    )

											# params
											for name, p in module.named_parameters(recurse=UpperCAmelCase_    ):
																		self.analyse_variable(UpperCAmelCase_							,			UpperCAmelCase_    )

											# inputs
											if isinstance(UpperCAmelCase_							,			UpperCAmelCase_    ):
																		for i, x in enumerate(UpperCAmelCase_    ):
																									self.analyse_variable(UpperCAmelCase_							,			F"""input[{i}]"""    )
											else:
																		self.analyse_variable(UpperCAmelCase_							,			'''input'''    )

											# outputs
											if isinstance(UpperCAmelCase_							,			UpperCAmelCase_    ):
																		for i, x in enumerate(UpperCAmelCase_    ):
																									# possibly a tuple of tuples
																									if isinstance(UpperCAmelCase_							,			UpperCAmelCase_    ):
																																for j, y in enumerate(UpperCAmelCase_    ):
																																							self.analyse_variable(UpperCAmelCase_							,			F"""output[{i}][{j}]"""    )
																									else:
																																self.analyse_variable(UpperCAmelCase_							,			F"""output[{i}]"""    )
											else:
																		self.analyse_variable(UpperCAmelCase_							,			'''output'''    )

											self.save_frame()



				def    __snake_case		(			self    ):
											self.model.apply(self._register_forward_hook    )



				def    __snake_case		(			self							,			UpperCAmelCase_    ):
											module.register_forward_hook(self.forward_hook    )






				def    __snake_case		(			self							,			UpperCAmelCase_							,			UpperCAmelCase_							,			UpperCAmelCase_    ):
											# - input is a tuple of packed inputs (could be non-Tensors)
											# - output could be a Tensor or a tuple of Tensors and non-Tensors

											lowerCAmelCase =		False

											lowerCAmelCase =		True if self.batch_number in self.trace_batch_nums else False
											if trace_mode:
																		self.reset_saved_frames()

											if self.total_calls == 0:
																		self.batch_start_frame()
											self.total_calls += 1

											# count batch numbers - the very first forward hook of the batch will be called when the
											# batch completes - i.e. it gets called very last - we know this batch has finished
											if module == self.model:
																		self.batch_number += 1
																		lowerCAmelCase =		True

											self.create_frame(UpperCAmelCase_							,			UpperCAmelCase_							,			UpperCAmelCase_    )

											# if last_frame_of_batch:
											#     self.batch_end_frame()

											if trace_mode:
																		self.trace_frames()

											if last_frame_of_batch:
																		self.batch_start_frame()

											if self.detected_overflow and not trace_mode:
																		self.dump_saved_frames()

																		# now we can abort, as it's pointless to continue running
																		raise ValueError(
																		    '''DebugUnderflowOverflow: inf/nan detected, aborting as there is no point running further. '''
																		    '''Please scroll up above this traceback to see the activation values prior to this event.'''    )

											# abort after certain batch if requested to do so
											if self.abort_after_batch_num is not None and self.batch_number > self.abort_after_batch_num:
																		raise ValueError(
																		    F"""DebugUnderflowOverflow: aborting after {self.batch_number} batches due to"""
																		    F""" `abort_after_batch_num={self.abort_after_batch_num}` arg"""    )




def    UpperCAmelCase							(  _snake_case    ,			_snake_case				):
							lowerCAmelCase =		var.abs()
							return F"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""



def    UpperCAmelCase							(  _snake_case    ,			_snake_case				):
							lowerCAmelCase =		False
							if torch.isnan(_snake_case				).any().item():
														lowerCAmelCase =		True
														print(F"""{ctx} has nans"""				)
							if torch.isinf(_snake_case				).any().item():
														lowerCAmelCase =		True
														print(F"""{ctx} has infs"""				)

							# if needed to monitor large elements can enable the following
							if 0:  # and detected:
														lowerCAmelCase =		var[torch.ge(var.abs()    ,			100				)]
														if naaa.numel() > 0:
																					print(F"""{ctx}:  n100={naaa.numel()}"""				)
														lowerCAmelCase =		var[torch.ge(var.abs()    ,			1000				)]
														if naaaa.numel() > 0:
																					print(F"""{ctx}: n1000={naaaa.numel()}"""				)
														lowerCAmelCase =		var[torch.ge(var.abs()    ,			10000				)]
														if naaaaa.numel() > 0:
																					print(F"""{ctx}: n10000={naaaaa.numel()}"""				)

							if 0:
														print(F"""min={var.min():9.2e} max={var.max():9.2e}"""				)

							if 0:
														print(F"""min={var.min():9.2e} max={var.max():9.2e} var={var.var():9.2e} mean={var.mean():9.2e} ({ctx})"""				)

							return detected


class 				__UpperCamelCase      (			__UpperCAmelCase	):





				'''simple docstring'''



				__a				:			Tuple					    ="""underflow_overflow"""
				__a				:			Dict					    ="""tpu_metrics_debug"""



