
"""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 				A__ (			_lowerCamelCase):
		A_		: torch.FloatTensor




class 				A__ (			nn.Module):

		def __init__(   self ,	_SCREAMING_SNAKE_CASE=3 ,	_SCREAMING_SNAKE_CASE=3 ,	_SCREAMING_SNAKE_CASE=("DownEncoderBlock2D",) ,	_SCREAMING_SNAKE_CASE=(64,) ,	_SCREAMING_SNAKE_CASE=2 ,	_SCREAMING_SNAKE_CASE=32 ,	_SCREAMING_SNAKE_CASE="silu" ,	_SCREAMING_SNAKE_CASE=True ,	):
			super().__init__()
			__lowerCAmelCase    :		Union[str, Any]		    =   layers_per_block

			__lowerCAmelCase    :		Dict		    =   torch.nn.Convad(
			    _SCREAMING_SNAKE_CASE ,	block_out_channels[0] ,	kernel_size=3 ,	stride=1 ,	padding=1 ,	)

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

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

				__lowerCAmelCase    :		Any		    =   get_down_block(
				    _SCREAMING_SNAKE_CASE ,	num_layers=self.layers_per_block ,	in_channels=_SCREAMING_SNAKE_CASE ,	out_channels=_SCREAMING_SNAKE_CASE ,	add_downsample=not is_final_block ,	resnet_eps=1E-6 ,	downsample_padding=0 ,	resnet_act_fn=_SCREAMING_SNAKE_CASE ,	resnet_groups=_SCREAMING_SNAKE_CASE ,	attention_head_dim=_SCREAMING_SNAKE_CASE ,	temb_channels=_SCREAMING_SNAKE_CASE ,	)
				self.down_blocks.append(_SCREAMING_SNAKE_CASE				)

			# mid
			__lowerCAmelCase    :		Optional[Any]		    =   UNetMidBlockaD(
			    in_channels=block_out_channels[-1] ,	resnet_eps=1E-6 ,	resnet_act_fn=_SCREAMING_SNAKE_CASE ,	output_scale_factor=1 ,	resnet_time_scale_shift='default' ,	attention_head_dim=block_out_channels[-1] ,	resnet_groups=_SCREAMING_SNAKE_CASE ,	temb_channels=_SCREAMING_SNAKE_CASE ,	)

			# out
			__lowerCAmelCase    :		List[Any]		    =   nn.GroupNorm(num_channels=block_out_channels[-1] ,	num_groups=_SCREAMING_SNAKE_CASE ,	eps=1E-6				)
			__lowerCAmelCase    :		Optional[int]		    =   nn.SiLU()

			__lowerCAmelCase    :		Union[str, Any]		    =   2 * out_channels if double_z else out_channels
			__lowerCAmelCase    :		Dict		    =   nn.Convad(block_out_channels[-1] ,	_SCREAMING_SNAKE_CASE ,	3 ,	padding=1				)

			__lowerCAmelCase    :		Any		    =   False







		def 							__lowerCamelCase							(   self ,	_SCREAMING_SNAKE_CASE				):
			__lowerCAmelCase    :		Dict		    =   x
			__lowerCAmelCase    :		List[str]		    =   self.conv_in(_SCREAMING_SNAKE_CASE				)

			if self.training and self.gradient_checkpointing:

				def create_custom_forward(_SCREAMING_SNAKE_CASE				):
					def custom_forward(*_SCREAMING_SNAKE_CASE				):
						return module(*_SCREAMING_SNAKE_CASE				)

					return custom_forward

				# down
				if is_torch_version('>=' ,	'1.11.0'				):
					for down_block in self.down_blocks:
						__lowerCAmelCase    :		List[Any]		    =   torch.utils.checkpoint.checkpoint(
						    create_custom_forward(_SCREAMING_SNAKE_CASE				) ,	_SCREAMING_SNAKE_CASE ,	use_reentrant=_SCREAMING_SNAKE_CASE				)
					# middle
					__lowerCAmelCase    :		str		    =   torch.utils.checkpoint.checkpoint(
					    create_custom_forward(self.mid_block				) ,	_SCREAMING_SNAKE_CASE ,	use_reentrant=_SCREAMING_SNAKE_CASE				)
				else:
					for down_block in self.down_blocks:
						__lowerCAmelCase    :		Optional[Any]		    =   torch.utils.checkpoint.checkpoint(create_custom_forward(_SCREAMING_SNAKE_CASE				) ,	_SCREAMING_SNAKE_CASE				)
					# middle
					__lowerCAmelCase    :		Tuple		    =   torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block				) ,	_SCREAMING_SNAKE_CASE				)

			else:
				# down
				for down_block in self.down_blocks:
					__lowerCAmelCase    :		int		    =   down_block(_SCREAMING_SNAKE_CASE				)

				# middle
				__lowerCAmelCase    :		Union[str, Any]		    =   self.mid_block(_SCREAMING_SNAKE_CASE				)

			# post-process
			__lowerCAmelCase    :		Union[str, Any]		    =   self.conv_norm_out(_SCREAMING_SNAKE_CASE				)
			__lowerCAmelCase    :		str		    =   self.conv_act(_SCREAMING_SNAKE_CASE				)
			__lowerCAmelCase    :		Optional[int]		    =   self.conv_out(_SCREAMING_SNAKE_CASE				)

			return sample




class 				A__ (			nn.Module):

		def __init__(   self ,	_SCREAMING_SNAKE_CASE=3 ,	_SCREAMING_SNAKE_CASE=3 ,	_SCREAMING_SNAKE_CASE=("UpDecoderBlock2D",) ,	_SCREAMING_SNAKE_CASE=(64,) ,	_SCREAMING_SNAKE_CASE=2 ,	_SCREAMING_SNAKE_CASE=32 ,	_SCREAMING_SNAKE_CASE="silu" ,	_SCREAMING_SNAKE_CASE="group" ,	):
			super().__init__()
			__lowerCAmelCase    :		Optional[int]		    =   layers_per_block

			__lowerCAmelCase    :		Optional[Any]		    =   nn.Convad(
			    _SCREAMING_SNAKE_CASE ,	block_out_channels[-1] ,	kernel_size=3 ,	stride=1 ,	padding=1 ,	)

			__lowerCAmelCase    :		List[str]		    =   None
			__lowerCAmelCase    :		List[str]		    =   nn.ModuleList([]				)

			__lowerCAmelCase    :		Union[str, Any]		    =   in_channels if norm_type == 'spatial' else None

			# mid
			__lowerCAmelCase    :		List[Any]		    =   UNetMidBlockaD(
			    in_channels=block_out_channels[-1] ,	resnet_eps=1E-6 ,	resnet_act_fn=_SCREAMING_SNAKE_CASE ,	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=_SCREAMING_SNAKE_CASE ,	temb_channels=_SCREAMING_SNAKE_CASE ,	)

			# up
			__lowerCAmelCase    :		Any		    =   list(reversed(_SCREAMING_SNAKE_CASE				)				)
			__lowerCAmelCase    :		Union[str, Any]		    =   reversed_block_out_channels[0]
			for i, up_block_type in enumerate(_SCREAMING_SNAKE_CASE				):
				__lowerCAmelCase    :		List[str]		    =   output_channel
				__lowerCAmelCase    :		Tuple		    =   reversed_block_out_channels[i]

				__lowerCAmelCase    :		Dict		    =   i == len(_SCREAMING_SNAKE_CASE				) - 1

				__lowerCAmelCase    :		Tuple		    =   get_up_block(
				    _SCREAMING_SNAKE_CASE ,	num_layers=self.layers_per_block + 1 ,	in_channels=_SCREAMING_SNAKE_CASE ,	out_channels=_SCREAMING_SNAKE_CASE ,	prev_output_channel=_SCREAMING_SNAKE_CASE ,	add_upsample=not is_final_block ,	resnet_eps=1E-6 ,	resnet_act_fn=_SCREAMING_SNAKE_CASE ,	resnet_groups=_SCREAMING_SNAKE_CASE ,	attention_head_dim=_SCREAMING_SNAKE_CASE ,	temb_channels=_SCREAMING_SNAKE_CASE ,	resnet_time_scale_shift=_SCREAMING_SNAKE_CASE ,	)
				self.up_blocks.append(_SCREAMING_SNAKE_CASE				)
				__lowerCAmelCase    :		List[str]		    =   output_channel

			# out
			if norm_type == "spatial":
				__lowerCAmelCase    :		Union[str, Any]		    =   SpatialNorm(block_out_channels[0] ,	_SCREAMING_SNAKE_CASE				)
			else:
				__lowerCAmelCase    :		str		    =   nn.GroupNorm(num_channels=block_out_channels[0] ,	num_groups=_SCREAMING_SNAKE_CASE ,	eps=1E-6				)
			__lowerCAmelCase    :		Dict		    =   nn.SiLU()
			__lowerCAmelCase    :		Union[str, Any]		    =   nn.Convad(block_out_channels[0] ,	_SCREAMING_SNAKE_CASE ,	3 ,	padding=1				)

			__lowerCAmelCase    :		Optional[Any]		    =   False







		def 							__lowerCamelCase							(   self ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE=None				):
			__lowerCAmelCase    :		str		    =   z
			__lowerCAmelCase    :		Optional[Any]		    =   self.conv_in(_SCREAMING_SNAKE_CASE				)

			__lowerCAmelCase    :		str		    =   next(iter(self.up_blocks.parameters()				)				).dtype
			if self.training and self.gradient_checkpointing:

				def create_custom_forward(_SCREAMING_SNAKE_CASE				):
					def custom_forward(*_SCREAMING_SNAKE_CASE				):
						return module(*_SCREAMING_SNAKE_CASE				)

					return custom_forward

				if is_torch_version('>=' ,	'1.11.0'				):
					# middle
					__lowerCAmelCase    :		List[Any]		    =   torch.utils.checkpoint.checkpoint(
					    create_custom_forward(self.mid_block				) ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	use_reentrant=_SCREAMING_SNAKE_CASE				)
					__lowerCAmelCase    :		Union[str, Any]		    =   sample.to(_SCREAMING_SNAKE_CASE				)

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

					# up
					for up_block in self.up_blocks:
						__lowerCAmelCase    :		Any		    =   torch.utils.checkpoint.checkpoint(create_custom_forward(_SCREAMING_SNAKE_CASE				) ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)
			else:
				# middle
				__lowerCAmelCase    :		str		    =   self.mid_block(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)
				__lowerCAmelCase    :		List[Any]		    =   sample.to(_SCREAMING_SNAKE_CASE				)

				# up
				for up_block in self.up_blocks:
					__lowerCAmelCase    :		Dict		    =   up_block(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)

        # post-process
			if latent_embeds is None:
				__lowerCAmelCase    :		Any		    =   self.conv_norm_out(_SCREAMING_SNAKE_CASE				)
			else:
				__lowerCAmelCase    :		int		    =   self.conv_norm_out(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)
			__lowerCAmelCase    :		Optional[Any]		    =   self.conv_act(_SCREAMING_SNAKE_CASE				)
			__lowerCAmelCase    :		Union[str, Any]		    =   self.conv_out(_SCREAMING_SNAKE_CASE				)

			return sample




class 				A__ (			nn.Module):

		def __init__(   self ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE=None ,	_SCREAMING_SNAKE_CASE="random" ,	_SCREAMING_SNAKE_CASE=False ,	_SCREAMING_SNAKE_CASE=True				):
			super().__init__()
			__lowerCAmelCase    :		Tuple		    =   n_e
			__lowerCAmelCase    :		Optional[int]		    =   vq_embed_dim
			__lowerCAmelCase    :		Union[str, Any]		    =   beta
			__lowerCAmelCase    :		List[str]		    =   legacy

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

			__lowerCAmelCase    :		Dict		    =   remap
			if self.remap is not None:
				self.register_buffer('used' ,	torch.tensor(np.load(self.remap				)				)				)
				__lowerCAmelCase    :		str		    =   self.used.shape[0]
				__lowerCAmelCase    :		Any		    =   unknown_index  # "random" or "extra" or integer
				if self.unknown_index == "extra":
					__lowerCAmelCase    :		Any		    =   self.re_embed
					__lowerCAmelCase    :		List[str]		    =   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:
				__lowerCAmelCase    :		Optional[int]		    =   n_e

			__lowerCAmelCase    :		Any		    =   sane_index_shape

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

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

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

			# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
			__lowerCAmelCase    :		List[str]		    =   torch.argmin(torch.cdist(_SCREAMING_SNAKE_CASE ,	self.embedding.weight				) ,	dim=1				)

			__lowerCAmelCase    :		str		    =   self.embedding(_SCREAMING_SNAKE_CASE				).view(z.shape				)
			__lowerCAmelCase    :		Optional[int]		    =   None
			__lowerCAmelCase    :		Union[str, Any]		    =   None

			# compute loss for embedding
			if not self.legacy:
				__lowerCAmelCase    :		Union[str, Any]		    =   self.beta * torch.mean((z_q.detach() - z) ** 2				) + torch.mean((z_q - z.detach()) ** 2				)
			else:
				__lowerCAmelCase    :		List[str]		    =   torch.mean((z_q.detach() - z) ** 2				) + self.beta * torch.mean((z_q - z.detach()) ** 2				)

			# preserve gradients
			__lowerCAmelCase    :		Dict		    =   z + (z_q - z).detach()

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

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

			if self.sane_index_shape:
				__lowerCAmelCase    :		Union[str, Any]		    =   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 ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				):
			# shape specifying (batch, height, width, channel)
			if self.remap is not None:
				__lowerCAmelCase    :		Tuple		    =   indices.reshape(shape[0] ,	-1				)  # add batch axis
				__lowerCAmelCase    :		List[Any]		    =   self.unmap_to_all(_SCREAMING_SNAKE_CASE				)
				__lowerCAmelCase    :		Any		    =   indices.reshape(-1				)  # flatten again

			# get quantized latent vectors
			__lowerCAmelCase    :		List[str]		    =   self.embedding(_SCREAMING_SNAKE_CASE				)

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

			return z_q




class 				A__ (			_lowerCamelCase):

		def __init__(   self ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE=False				):
			__lowerCAmelCase    :		List[Any]		    =   parameters
			__lowerCAmelCase    ,     __lowerCAmelCase    :		int		    =   torch.chunk(_SCREAMING_SNAKE_CASE ,	2 ,	dim=1				)
			__lowerCAmelCase    :		int		    =   torch.clamp(self.logvar ,	-30.0 ,	20.0				)
			__lowerCAmelCase    :		List[str]		    =   deterministic
			__lowerCAmelCase    :		Optional[int]		    =   torch.exp(0.5 * self.logvar				)
			__lowerCAmelCase    :		str		    =   torch.exp(self.logvar				)
			if self.deterministic:
				__lowerCAmelCase    :		Optional[int]		    =   torch.zeros_like(
				    self.mean ,	device=self.parameters.device ,	dtype=self.parameters.dtype				)

		def 							__lowerCamelCase							(   self ,	_SCREAMING_SNAKE_CASE = None				):
			# make sure sample is on the same device as the parameters and has same dtype
			__lowerCAmelCase    :		Tuple		    =   randn_tensor(
			    self.mean.shape ,	generator=_SCREAMING_SNAKE_CASE ,	device=self.parameters.device ,	dtype=self.parameters.dtype				)
			__lowerCAmelCase    :		Dict		    =   self.mean + self.std * sample
			return x

		def 							__lowerCamelCase							(   self ,	_SCREAMING_SNAKE_CASE=None				):
			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 ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE=[1, 2, 3]				):
			if self.deterministic:
				return torch.Tensor([0.0]				)
			__lowerCAmelCase    :		Any		    =   np.log(2.0 * np.pi				)
			return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean ,	2				) / self.var ,	dim=_SCREAMING_SNAKE_CASE				)







		def 							__lowerCamelCase							(   self				):
			return self.mean

"""simple docstring"""







def 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase	):

	if density <= 0:
		raise ValueError('Impossible fluid density'	)
	if bulk_modulus <= 0:
		raise ValueError('Impossible bulk modulus'	)

	return (bulk_modulus / density) ** 0.5


if __name__ == "__main__":
							import doctest

							doctest.testmod()