






"""simple docstring"""
from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block







@dataclass
class 			UpperCamelCase_       (__A     ):
	__magic_name__	      =				42







class 			UpperCamelCase_       (nn.Module     ):





	def __init__(      self					: Optional[int]	,	lowerCAmelCase_					: List[str]=3	,	lowerCAmelCase_					: int=3	,	lowerCAmelCase_					: List[str]=("DownEncoderBlock2D",)	,	lowerCAmelCase_					: str=(64,)	,	lowerCAmelCase_					: Union[str, Any]=2	,	lowerCAmelCase_					: Any=32	,	lowerCAmelCase_					: str="silu"	,	lowerCAmelCase_					: Optional[int]=True	,	)						->			List[Any]:
				super().__init__()
				UpperCAmelCase_     : Optional[int]			     =					layers_per_block

				UpperCAmelCase_     : Tuple			     =					torch.nn.Convad(
				    lowerCAmelCase_	,	block_out_channels[0]	,	kernel_size=3	,	stride=1	,	padding=1	,	)

				UpperCAmelCase_     : List[Any]			     =					None
				UpperCAmelCase_     : Optional[Any]			     =					nn.ModuleList([]						)

				# down
				UpperCAmelCase_     : List[str]			     =					block_out_channels[0]
				for i, down_block_type in enumerate(lowerCAmelCase_						):
							UpperCAmelCase_     : Union[str, Any]			     =					output_channel
							UpperCAmelCase_     : Dict			     =					block_out_channels[i]
							UpperCAmelCase_     : Optional[int]			     =					i == len(lowerCAmelCase_						) - 1

							UpperCAmelCase_     : Optional[int]			     =					get_down_block(
							    lowerCAmelCase_	,	num_layers=self.layers_per_block	,	in_channels=lowerCAmelCase_	,	out_channels=lowerCAmelCase_	,	add_downsample=not is_final_block	,	resnet_eps=1e-6	,	downsample_padding=0	,	resnet_act_fn=lowerCAmelCase_	,	resnet_groups=lowerCAmelCase_	,	attention_head_dim=lowerCAmelCase_	,	temb_channels=lowerCAmelCase_	,	)
							self.down_blocks.append(lowerCAmelCase_						)

				# mid
				UpperCAmelCase_     : Any			     =					UNetMidBlockaD(
				    in_channels=block_out_channels[-1]	,	resnet_eps=1e-6	,	resnet_act_fn=lowerCAmelCase_	,	output_scale_factor=1	,	resnet_time_scale_shift="default"	,	attention_head_dim=block_out_channels[-1]	,	resnet_groups=lowerCAmelCase_	,	temb_channels=lowerCAmelCase_	,	)

				# out
				UpperCAmelCase_     : int			     =					nn.GroupNorm(num_channels=block_out_channels[-1]	,	num_groups=lowerCAmelCase_	,	eps=1e-6						)
				UpperCAmelCase_     : List[str]			     =					nn.SiLU()

				UpperCAmelCase_     : Any			     =					2 * out_channels if double_z else out_channels
				UpperCAmelCase_     : List[Any]			     =					nn.Convad(block_out_channels[-1]	,	lowerCAmelCase_	,	3	,	padding=1						)

				UpperCAmelCase_     : Optional[int]			     =					False





	def      _SCREAMING_SNAKE_CASE      (      self					: Optional[int]	,	lowerCAmelCase_					: int						)						->			str:
				UpperCAmelCase_     : Dict			     =					x
				UpperCAmelCase_     : int			     =					self.conv_in(lowerCAmelCase_						)

				if self.training and self.gradient_checkpointing:

							def create_custom_forward(lowerCAmelCase_					: Optional[Any]						):
										def custom_forward(*lowerCAmelCase_					: Optional[Any]						):
													return module(*lowerCAmelCase_						)

										return custom_forward

							# down
							if is_torch_version(">="	,	"1.11.0"						):
										for down_block in self.down_blocks:
													UpperCAmelCase_     : Dict			     =					torch.utils.checkpoint.checkpoint(
													    create_custom_forward(lowerCAmelCase_						)	,	lowerCAmelCase_	,	use_reentrant=lowerCAmelCase_						)
										# middle
										UpperCAmelCase_     : Union[str, Any]			     =					torch.utils.checkpoint.checkpoint(
										    create_custom_forward(self.mid_block						)	,	lowerCAmelCase_	,	use_reentrant=lowerCAmelCase_						)
							else:
										for down_block in self.down_blocks:
													UpperCAmelCase_     : Tuple			     =					torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCAmelCase_						)	,	lowerCAmelCase_						)
										# middle
										UpperCAmelCase_     : List[Any]			     =					torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block						)	,	lowerCAmelCase_						)

				else:
							# down
							for down_block in self.down_blocks:
										UpperCAmelCase_     : Optional[Any]			     =					down_block(lowerCAmelCase_						)

							# middle
							UpperCAmelCase_     : Tuple			     =					self.mid_block(lowerCAmelCase_						)

				# post-process
				UpperCAmelCase_     : List[str]			     =					self.conv_norm_out(lowerCAmelCase_						)
				UpperCAmelCase_     : str			     =					self.conv_act(lowerCAmelCase_						)
				UpperCAmelCase_     : Optional[int]			     =					self.conv_out(lowerCAmelCase_						)

				return sample







class 			UpperCamelCase_       (nn.Module     ):





	def __init__(      self					: str	,	lowerCAmelCase_					: Union[str, Any]=3	,	lowerCAmelCase_					: Any=3	,	lowerCAmelCase_					: Optional[int]=("UpDecoderBlock2D",)	,	lowerCAmelCase_					: Dict=(64,)	,	lowerCAmelCase_					: Union[str, Any]=2	,	lowerCAmelCase_					: Tuple=32	,	lowerCAmelCase_					: Dict="silu"	,	lowerCAmelCase_					: Union[str, Any]="group"	,	)						->			int:
				super().__init__()
				UpperCAmelCase_     : Optional[int]			     =					layers_per_block

				UpperCAmelCase_     : List[Any]			     =					nn.Convad(
				    lowerCAmelCase_	,	block_out_channels[-1]	,	kernel_size=3	,	stride=1	,	padding=1	,	)

				UpperCAmelCase_     : str			     =					None
				UpperCAmelCase_     : Optional[Any]			     =					nn.ModuleList([]						)

				UpperCAmelCase_     : List[str]			     =					in_channels if norm_type == "spatial" else None

				# mid
				UpperCAmelCase_     : List[str]			     =					UNetMidBlockaD(
				    in_channels=block_out_channels[-1]	,	resnet_eps=1e-6	,	resnet_act_fn=lowerCAmelCase_	,	output_scale_factor=1	,	resnet_time_scale_shift="default" if norm_type == "group" else norm_type	,	attention_head_dim=block_out_channels[-1]	,	resnet_groups=lowerCAmelCase_	,	temb_channels=lowerCAmelCase_	,	)

				# up
				UpperCAmelCase_     : Optional[int]			     =					list(reversed(lowerCAmelCase_						)						)
				UpperCAmelCase_     : Tuple			     =					reversed_block_out_channels[0]
				for i, up_block_type in enumerate(lowerCAmelCase_						):
							UpperCAmelCase_     : Optional[int]			     =					output_channel
							UpperCAmelCase_     : int			     =					reversed_block_out_channels[i]

							UpperCAmelCase_     : str			     =					i == len(lowerCAmelCase_						) - 1

							UpperCAmelCase_     : int			     =					get_up_block(
							    lowerCAmelCase_	,	num_layers=self.layers_per_block + 1	,	in_channels=lowerCAmelCase_	,	out_channels=lowerCAmelCase_	,	prev_output_channel=lowerCAmelCase_	,	add_upsample=not is_final_block	,	resnet_eps=1e-6	,	resnet_act_fn=lowerCAmelCase_	,	resnet_groups=lowerCAmelCase_	,	attention_head_dim=lowerCAmelCase_	,	temb_channels=lowerCAmelCase_	,	resnet_time_scale_shift=lowerCAmelCase_	,	)
							self.up_blocks.append(lowerCAmelCase_						)
							UpperCAmelCase_     : str			     =					output_channel

				# out
				if norm_type == "spatial":
							UpperCAmelCase_     : List[str]			     =					SpatialNorm(block_out_channels[0]	,	lowerCAmelCase_						)
				else:
							UpperCAmelCase_     : List[str]			     =					nn.GroupNorm(num_channels=block_out_channels[0]	,	num_groups=lowerCAmelCase_	,	eps=1e-6						)
				UpperCAmelCase_     : List[Any]			     =					nn.SiLU()
				UpperCAmelCase_     : Optional[Any]			     =					nn.Convad(block_out_channels[0]	,	lowerCAmelCase_	,	3	,	padding=1						)

				UpperCAmelCase_     : List[str]			     =					False





	def      _SCREAMING_SNAKE_CASE      (      self					: List[str]	,	lowerCAmelCase_					: str	,	lowerCAmelCase_					: Dict=None						)						->			List[Any]:
				UpperCAmelCase_     : Optional[Any]			     =					z
				UpperCAmelCase_     : Any			     =					self.conv_in(lowerCAmelCase_						)

				UpperCAmelCase_     : List[str]			     =					next(iter(self.up_blocks.parameters()						)						).dtype
				if self.training and self.gradient_checkpointing:

							def create_custom_forward(lowerCAmelCase_					: Optional[int]						):
										def custom_forward(*lowerCAmelCase_					: Optional[int]						):
													return module(*lowerCAmelCase_						)

										return custom_forward

							if is_torch_version(">="	,	"1.11.0"						):
										# middle
										UpperCAmelCase_     : List[str]			     =					torch.utils.checkpoint.checkpoint(
										    create_custom_forward(self.mid_block						)	,	lowerCAmelCase_	,	lowerCAmelCase_	,	use_reentrant=lowerCAmelCase_						)
										UpperCAmelCase_     : Optional[Any]			     =					sample.to(lowerCAmelCase_						)

										# up
										for up_block in self.up_blocks:
													UpperCAmelCase_     : Optional[int]			     =					torch.utils.checkpoint.checkpoint(
													    create_custom_forward(lowerCAmelCase_						)	,	lowerCAmelCase_	,	lowerCAmelCase_	,	use_reentrant=lowerCAmelCase_						)
							else:
										# middle
										UpperCAmelCase_     : List[str]			     =					torch.utils.checkpoint.checkpoint(
										    create_custom_forward(self.mid_block						)	,	lowerCAmelCase_	,	lowerCAmelCase_						)
										UpperCAmelCase_     : Any			     =					sample.to(lowerCAmelCase_						)

										# up
										for up_block in self.up_blocks:
													UpperCAmelCase_     : int			     =					torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCAmelCase_						)	,	lowerCAmelCase_	,	lowerCAmelCase_						)
				else:
							# middle
							UpperCAmelCase_     : Optional[Any]			     =					self.mid_block(lowerCAmelCase_	,	lowerCAmelCase_						)
							UpperCAmelCase_     : Optional[Any]			     =					sample.to(lowerCAmelCase_						)

							# up
							for up_block in self.up_blocks:
										UpperCAmelCase_     : Tuple			     =					up_block(lowerCAmelCase_	,	lowerCAmelCase_						)

        # post-process
				if latent_embeds is None:
							UpperCAmelCase_     : Optional[int]			     =					self.conv_norm_out(lowerCAmelCase_						)
				else:
							UpperCAmelCase_     : Tuple			     =					self.conv_norm_out(lowerCAmelCase_	,	lowerCAmelCase_						)
				UpperCAmelCase_     : Optional[Any]			     =					self.conv_act(lowerCAmelCase_						)
				UpperCAmelCase_     : List[str]			     =					self.conv_out(lowerCAmelCase_						)

				return sample







class 			UpperCamelCase_       (nn.Module     ):





	def __init__(      self					: Optional[Any]	,	lowerCAmelCase_					: Optional[Any]	,	lowerCAmelCase_					: Any	,	lowerCAmelCase_					: List[str]	,	lowerCAmelCase_					: Optional[int]=None	,	lowerCAmelCase_					: List[Any]="random"	,	lowerCAmelCase_					: Union[str, Any]=False	,	lowerCAmelCase_					: str=True						)						->			Any:
				super().__init__()
				UpperCAmelCase_     : List[str]			     =					n_e
				UpperCAmelCase_     : List[Any]			     =					vq_embed_dim
				UpperCAmelCase_     : List[str]			     =					beta
				UpperCAmelCase_     : Union[str, Any]			     =					legacy

				UpperCAmelCase_     : Tuple			     =					nn.Embedding(self.n_e	,	self.vq_embed_dim						)
				self.embedding.weight.data.uniform_(-1.0 / self.n_e	,	1.0 / self.n_e						)

				UpperCAmelCase_     : List[str]			     =					remap
				if self.remap is not None:
							self.register_buffer("used"	,	torch.tensor(np.load(self.remap						)						)						)
							UpperCAmelCase_     : List[Any]			     =					self.used.shape[0]
							UpperCAmelCase_     : Optional[Any]			     =					unknown_index  # "random" or "extra" or integer
							if self.unknown_index == "extra":
										UpperCAmelCase_     : Union[str, Any]			     =					self.re_embed
										UpperCAmelCase_     : Optional[Any]			     =					self.re_embed + 1
							print(
							    f"""Remapping {self.n_e} indices to {self.re_embed} indices. """
							    f"""Using {self.unknown_index} for unknown indices."""						)
				else:
							UpperCAmelCase_     : List[str]			     =					n_e

				UpperCAmelCase_     : Optional[Any]			     =					sane_index_shape





	def      _SCREAMING_SNAKE_CASE      (      self					: Tuple	,	lowerCAmelCase_					: List[Any]						)						->			Any:
				UpperCAmelCase_     : str			     =					inds.shape
				assert len(lowerCAmelCase_						) > 1
				UpperCAmelCase_     : int			     =					inds.reshape(ishape[0]	,	-1						)
				UpperCAmelCase_     : Tuple			     =					self.used.to(lowerCAmelCase_						)
				UpperCAmelCase_     : Any			     =					(inds[:, :, None] == used[None, None, ...]).long()
				UpperCAmelCase_     : Optional[Any]			     =					match.argmax(-1						)
				UpperCAmelCase_     : Optional[Any]			     =					match.sum(2						) < 1
				if self.unknown_index == "random":
							UpperCAmelCase_     : List[str]			     =					torch.randint(0	,	self.re_embed	,	size=new[unknown].shape						).to(device=new.device						)
				else:
							UpperCAmelCase_     : List[str]			     =					self.unknown_index
				return new.reshape(lowerCAmelCase_						)





	def      _SCREAMING_SNAKE_CASE      (      self					: Any	,	lowerCAmelCase_					: Tuple						)						->			Any:
				UpperCAmelCase_     : str			     =					inds.shape
				assert len(lowerCAmelCase_						) > 1
				UpperCAmelCase_     : Dict			     =					inds.reshape(ishape[0]	,	-1						)
				UpperCAmelCase_     : str			     =					self.used.to(lowerCAmelCase_						)
				if self.re_embed > self.used.shape[0]:  # extra token
							UpperCAmelCase_     : Union[str, Any]			     =					0  # simply set to zero
				UpperCAmelCase_     : Tuple			     =					torch.gather(used[None, :][inds.shape[0] * [0], :]	,	1	,	lowerCAmelCase_						)
				return back.reshape(lowerCAmelCase_						)





	def      _SCREAMING_SNAKE_CASE      (      self					: Dict	,	lowerCAmelCase_					: Union[str, Any]						)						->			Optional[Any]:
				# reshape z -> (batch, height, width, channel) and flatten
				UpperCAmelCase_     : Union[str, Any]			     =					z.permute(0	,	2	,	3	,	1						).contiguous()
				UpperCAmelCase_     : Optional[Any]			     =					z.view(-1	,	self.vq_embed_dim						)

				# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
				UpperCAmelCase_     : Optional[Any]			     =					torch.argmin(torch.cdist(lowerCAmelCase_	,	self.embedding.weight						)	,	dim=1						)

				UpperCAmelCase_     : List[Any]			     =					self.embedding(lowerCAmelCase_						).view(z.shape						)
				UpperCAmelCase_     : List[str]			     =					None
				UpperCAmelCase_     : List[Any]			     =					None

				# compute loss for embedding
				if not self.legacy:
							UpperCAmelCase_     : Optional[Any]			     =					self.beta * torch.mean((z_q.detach() - z) ** 2						) + torch.mean((z_q - z.detach()) ** 2						)
				else:
							UpperCAmelCase_     : Optional[int]			     =					torch.mean((z_q.detach() - z) ** 2						) + self.beta * torch.mean((z_q - z.detach()) ** 2						)

				# preserve gradients
				UpperCAmelCase_     : str			     =					z + (z_q - z).detach()

				# reshape back to match original input shape
				UpperCAmelCase_     : Union[str, Any]			     =					z_q.permute(0	,	3	,	1	,	2						).contiguous()

				if self.remap is not None:
							UpperCAmelCase_     : int			     =					min_encoding_indices.reshape(z.shape[0]	,	-1						)  # add batch axis
							UpperCAmelCase_     : str			     =					self.remap_to_used(lowerCAmelCase_						)
							UpperCAmelCase_     : List[str]			     =					min_encoding_indices.reshape(-1	,	1						)  # flatten

				if self.sane_index_shape:
							UpperCAmelCase_     : List[str]			     =					min_encoding_indices.reshape(z_q.shape[0]	,	z_q.shape[2]	,	z_q.shape[3]						)

				return z_q, loss, (perplexity, min_encodings, min_encoding_indices)





	def      _SCREAMING_SNAKE_CASE      (      self					: List[Any]	,	lowerCAmelCase_					: Dict	,	lowerCAmelCase_					: Union[str, Any]						)						->			Optional[int]:
				# shape specifying (batch, height, width, channel)
				if self.remap is not None:
							UpperCAmelCase_     : Optional[Any]			     =					indices.reshape(shape[0]	,	-1						)  # add batch axis
							UpperCAmelCase_     : str			     =					self.unmap_to_all(lowerCAmelCase_						)
							UpperCAmelCase_     : Optional[Any]			     =					indices.reshape(-1						)  # flatten again

				# get quantized latent vectors
				UpperCAmelCase_     : Any			     =					self.embedding(lowerCAmelCase_						)

				if shape is not None:
							UpperCAmelCase_     : List[Any]			     =					z_q.view(lowerCAmelCase_						)
							# reshape back to match original input shape
							UpperCAmelCase_     : int			     =					z_q.permute(0	,	3	,	1	,	2						).contiguous()

				return z_q







class 			UpperCamelCase_       (__A     ):





	def __init__(      self					: Optional[Any]	,	lowerCAmelCase_					: str	,	lowerCAmelCase_					: Optional[Any]=False						)						->			List[Any]:
				UpperCAmelCase_     : int			     =					parameters
				UpperCAmelCase_     ,	UpperCAmelCase_     : Any			     =					torch.chunk(lowerCAmelCase_	,	2	,	dim=1						)
				UpperCAmelCase_     : Dict			     =					torch.clamp(self.logvar	,	-3_0.0	,	2_0.0						)
				UpperCAmelCase_     : List[Any]			     =					deterministic
				UpperCAmelCase_     : Tuple			     =					torch.exp(0.5 * self.logvar						)
				UpperCAmelCase_     : List[str]			     =					torch.exp(self.logvar						)
				if self.deterministic:
							UpperCAmelCase_     : List[str]			     =					torch.zeros_like(
							    self.mean	,	device=self.parameters.device	,	dtype=self.parameters.dtype						)





	def      _SCREAMING_SNAKE_CASE      (      self					: str	,	lowerCAmelCase_					: Optional[torch.Generator] = None						)						->			torch.FloatTensor:
				# make sure sample is on the same device as the parameters and has same dtype
				UpperCAmelCase_     : int			     =					randn_tensor(
				    self.mean.shape	,	generator=lowerCAmelCase_	,	device=self.parameters.device	,	dtype=self.parameters.dtype						)
				UpperCAmelCase_     : Optional[int]			     =					self.mean + self.std * sample
				return x





	def      _SCREAMING_SNAKE_CASE      (      self					: List[Any]	,	lowerCAmelCase_					: Optional[Any]=None						)						->			List[Any]:
				if self.deterministic:
							return torch.Tensor([0.0]						)
				else:
							if other is None:
										return 0.5 * torch.sum(torch.pow(self.mean	,	2						) + self.var - 1.0 - self.logvar	,	dim=[1, 2, 3]						)
							else:
										return 0.5 * torch.sum(
										    torch.pow(self.mean - other.mean	,	2						) / other.var
										    + self.var / other.var
										    - 1.0
										    - self.logvar
										    + other.logvar	,	dim=[1, 2, 3]	,	)





	def      _SCREAMING_SNAKE_CASE      (      self					: Any	,	lowerCAmelCase_					: List[str]	,	lowerCAmelCase_					: str=[1, 2, 3]						)						->			Optional[Any]:
				if self.deterministic:
							return torch.Tensor([0.0]						)
				UpperCAmelCase_     : str			     =					np.log(2.0 * np.pi						)
				return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean	,	2						) / self.var	,	dim=lowerCAmelCase_						)





	def      _SCREAMING_SNAKE_CASE      (      self					: Optional[int]						)						->			int:
				return self.mean











"""simple docstring"""
def 				snake_case      (   A__	,A__						):
			# "extended trapezoidal rule"
			# int(f) = dx/2 * (f1 + 2f2 + ... + fn)
			UpperCAmelCase_     : Dict			     =					(boundary[1] - boundary[0]) / steps
			UpperCAmelCase_     : Optional[int]			     =					boundary[0]
			UpperCAmelCase_     : str			     =					boundary[1]
			UpperCAmelCase_     : Tuple			     =					make_points(A__	,A__	,A__						)
			UpperCAmelCase_     : List[str]			     =					0.0
			y += (h / 2.0) * f(A__						)
			for i in x_i:
						# print(i)
						y += h * f(A__						)
			y += (h / 2.0) * f(A__						)
			return y

def 				snake_case      (   A__	,A__	,A__						):
			UpperCAmelCase_     : Union[str, Any]			     =					a + h
			while x < (b - h):
						yield x
						UpperCAmelCase_     : Optional[Any]			     =					x + h

def 				snake_case      (   A__						):  # enter your function here
			UpperCAmelCase_     : Dict			     =					(x - 0) * (x - 0)
			return y

def 				snake_case      (   ):
			UpperCAmelCase_     : Dict			     =					0.0  # Lower bound of integration
			UpperCAmelCase_     : Optional[int]			     =					1.0  # Upper bound of integration
			UpperCAmelCase_     : Dict			     =					10.0  # define number of steps or resolution
			UpperCAmelCase_     : List[Any]			     =					[a, b]  # define boundary of integration
			UpperCAmelCase_     : Union[str, Any]			     =					method_a(A__	,A__						)
			print(F"""y = {y}"""						)


if __name__ == "__main__":
			main()



