
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  lowerCamelCase__							(				__A):




	'''simple docstring'''




	_A								=     4_2




class  lowerCamelCase__							(				nn.Module):




	'''simple docstring'''




	def __init__(					self		:Optional[int]     ,   a		:List[str]=3     ,   a		:int=3     ,   a		:List[str]=("DownEncoderBlock2D",)     ,   a		:str=(6_4,)     ,   a		:Union[str, Any]=2     ,   a		:Any=3_2     ,   a		:str="silu"     ,   a		: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    _lowerCamelCase				(					self		:Optional[int]     ,   a		:int						)		-> str:
								__UpperCamelCase	:   Dict			   =					x
								__UpperCamelCase	:   int			   =					self.conv_in(lowerCAmelCase_						)

								if self.training and self.gradient_checkpointing:

															def create_custom_forward(a		:Optional[Any]						):
																						def custom_forward(*a		: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  lowerCamelCase__							(				nn.Module):




	'''simple docstring'''




	def __init__(					self		:str     ,   a		:Union[str, Any]=3     ,   a		:Any=3     ,   a		:Optional[int]=("UpDecoderBlock2D",)     ,   a		:Dict=(6_4,)     ,   a		:Union[str, Any]=2     ,   a		:Tuple=3_2     ,   a		:Dict="silu"     ,   a		: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    _lowerCamelCase				(					self		:List[str]     ,   a		:str     ,   a		: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(a		:Optional[int]						):
																						def custom_forward(*a		: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  lowerCamelCase__							(				nn.Module):




	'''simple docstring'''




	def __init__(					self		:Optional[Any]     ,   a		:Optional[Any]     ,   a		:Any     ,   a		:List[str]     ,   a		:Optional[int]=None     ,   a		:List[Any]="random"     ,   a		:Union[str, Any]=False     ,   a		: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    _lowerCamelCase				(					self		:Tuple     ,   a		: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    _lowerCamelCase				(					self		:Any     ,   a		: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    _lowerCamelCase				(					self		:Dict     ,   a		: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    _lowerCamelCase				(					self		:List[Any]     ,   a		:Dict     ,   a		: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  lowerCamelCase__							(				__A):




	'''simple docstring'''




	def __init__(					self		:Optional[Any]     ,   a		:str     ,   a		:Optional[Any]=False						)		-> List[Any]:
								__UpperCamelCase	:   int			   =					parameters
								__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    _lowerCamelCase				(					self		:str     ,   a		: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    _lowerCamelCase				(					self		:List[Any]     ,   a		: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    _lowerCamelCase				(					self		:Any     ,   a		:List[str]     ,   a		: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    _lowerCamelCase				(					self		:Optional[int]						)		-> int:
								return self.mean







"""simple docstring"""
lowerCamelCase_					=      [
    (1000, '''M'''),
    (900, '''CM'''),
    (500, '''D'''),
    (400, '''CD'''),
    (100, '''C'''),
    (90, '''XC'''),
    (50, '''L'''),
    (40, '''XL'''),
    (10, '''X'''),
    (9, '''IX'''),
    (5, '''V'''),
    (4, '''IV'''),
    (1, '''I'''),
]

def 				snake_case      (   A__						):
			UpperCAmelCase_     : List[str]			     =					{"I": 1, "V": 5, "X": 10, "L": 50, "C": 1_00, "D": 5_00, "M": 10_00}
			UpperCAmelCase_     : Optional[Any]			     =					0
			UpperCAmelCase_     : Tuple			     =					0
			while place < len(A__						):
						if (place + 1 < len(A__						)) and (vals[roman[place]] < vals[roman[place + 1]]):
									total += vals[roman[place + 1]] - vals[roman[place]]
									place += 2
						else:
									total += vals[roman[place]]
									place += 1
			return total

def 				snake_case      (   A__						):
			UpperCAmelCase_     : Union[str, Any]			     =					[]
			for arabic, roman in ROMAN:
						((UpperCAmelCase_)     ,	(UpperCAmelCase_))     : str			     =					divmod(A__	,A__						)
						result.append(roman * factor						)
						if number == 0:
									break
			return "".join(A__						)


if __name__ == "__main__":
			import doctest

			doctest.testmod()



