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 						SCREAMING_SNAKE_CASE				( SCREAMING_SNAKE_CASE__					):





							"""simple docstring"""

							A_						  =					42





class 						SCREAMING_SNAKE_CASE				( nn.Module					):





							"""simple docstring"""

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

												_A								= torch.nn.Convad(
												    snake_case__		,     block_out_channels[0]		,     kernel_size=3		,     stride=1		,     padding=1		,     )

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

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

																	_A								= get_down_block(
																	    snake_case__		,     num_layers=self.layers_per_block		,     in_channels=snake_case__		,     out_channels=snake_case__		,     add_downsample=not is_final_block		,     resnet_eps=1e-6		,     downsample_padding=0		,     resnet_act_fn=snake_case__		,     resnet_groups=snake_case__		,     attention_head_dim=snake_case__		,     temb_channels=snake_case__		,     )
																	self.down_blocks.append(snake_case__   )

												# mid
												_A								= UNetMidBlockaD(
												    in_channels=block_out_channels[-1]		,     resnet_eps=1e-6		,     resnet_act_fn=snake_case__		,     output_scale_factor=1		,     resnet_time_scale_shift='''default'''		,     attention_head_dim=block_out_channels[-1]		,     resnet_groups=snake_case__		,     temb_channels=snake_case__		,     )

												# out
												_A								= nn.GroupNorm(num_channels=block_out_channels[-1]		,     num_groups=snake_case__		,     eps=1e-6   )
												_A								= nn.SiLU()

												_A								= 2 * out_channels if double_z else out_channels
												_A								= nn.Convad(block_out_channels[-1]		,     snake_case__		,     3		,     padding=1   )

												_A								= False



							def       __A (  self:   Optional[int]		,     __A:   int   )    ->				Optional[int]:
												_A								= x
												_A								= self.conv_in(snake_case__   )

												if self.training and self.gradient_checkpointing:

																	def create_custom_forward(__A:   Dict   ):
																						def custom_forward(*__A:   Any   ):
																											return module(*snake_case__   )

																						return custom_forward

																	# down
																	if is_torch_version('''>='''		,     '''1.11.0'''   ):
																						for down_block in self.down_blocks:
																											_A								= torch.utils.checkpoint.checkpoint(
																											    create_custom_forward(snake_case__   )		,     snake_case__		,     use_reentrant=snake_case__   )
																						# middle
																						_A								= torch.utils.checkpoint.checkpoint(
																						    create_custom_forward(self.mid_block   )		,     snake_case__		,     use_reentrant=snake_case__   )
																	else:
																						for down_block in self.down_blocks:
																											_A								= torch.utils.checkpoint.checkpoint(create_custom_forward(snake_case__   )		,     snake_case__   )
																						# middle
																						_A								= torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block   )		,     snake_case__   )

												else:
																	# down
																	for down_block in self.down_blocks:
																						_A								= down_block(snake_case__   )

																	# middle
																	_A								= self.mid_block(snake_case__   )

												# post-process
												_A								= self.conv_norm_out(snake_case__   )
												_A								= self.conv_act(snake_case__   )
												_A								= self.conv_out(snake_case__   )

												return sample





class 						SCREAMING_SNAKE_CASE				( nn.Module					):





							"""simple docstring"""

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

												_A								= nn.Convad(
												    snake_case__		,     block_out_channels[-1]		,     kernel_size=3		,     stride=1		,     padding=1		,     )

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

												_A								= in_channels if norm_type == "spatial" else None

												# mid
												_A								= UNetMidBlockaD(
												    in_channels=block_out_channels[-1]		,     resnet_eps=1e-6		,     resnet_act_fn=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=snake_case__		,     temb_channels=snake_case__		,     )

												# up
												_A								= list(reversed(snake_case__   )   )
												_A								= reversed_block_out_channels[0]
												for i, up_block_type in enumerate(snake_case__   ):
																	_A								= output_channel
																	_A								= reversed_block_out_channels[i]

																	_A								= i == len(snake_case__   ) - 1

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

												# out
												if norm_type == "spatial":
																	_A								= SpatialNorm(block_out_channels[0]		,     snake_case__   )
												else:
																	_A								= nn.GroupNorm(num_channels=block_out_channels[0]		,     num_groups=snake_case__		,     eps=1e-6   )
												_A								= nn.SiLU()
												_A								= nn.Convad(block_out_channels[0]		,     snake_case__		,     3		,     padding=1   )

												_A								= False



							def       __A (  self:   Optional[Any]		,     __A:   Any		,     __A:   Any=None   )    ->				Union[str, Any]:
												_A								= z
												_A								= self.conv_in(snake_case__   )

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

																	def create_custom_forward(__A:   Dict   ):
																						def custom_forward(*__A:   Any   ):
																											return module(*snake_case__   )

																						return custom_forward

																	if is_torch_version('''>='''		,     '''1.11.0'''   ):
																						# middle
																						_A								= torch.utils.checkpoint.checkpoint(
																						    create_custom_forward(self.mid_block   )		,     snake_case__		,     snake_case__		,     use_reentrant=snake_case__   )
																						_A								= sample.to(snake_case__   )

																						# up
																						for up_block in self.up_blocks:
																											_A								= torch.utils.checkpoint.checkpoint(
																											    create_custom_forward(snake_case__   )		,     snake_case__		,     snake_case__		,     use_reentrant=snake_case__   )
																	else:
																						# middle
																						_A								= torch.utils.checkpoint.checkpoint(
																						    create_custom_forward(self.mid_block   )		,     snake_case__		,     snake_case__   )
																						_A								= sample.to(snake_case__   )

																						# up
																						for up_block in self.up_blocks:
																											_A								= torch.utils.checkpoint.checkpoint(create_custom_forward(snake_case__   )		,     snake_case__		,     snake_case__   )
												else:
																	# middle
																	_A								= self.mid_block(snake_case__		,     snake_case__   )
																	_A								= sample.to(snake_case__   )

																	# up
																	for up_block in self.up_blocks:
																						_A								= up_block(snake_case__		,     snake_case__   )

        # post-process
												if latent_embeds is None:
																	_A								= self.conv_norm_out(snake_case__   )
												else:
																	_A								= self.conv_norm_out(snake_case__		,     snake_case__   )
												_A								= self.conv_act(snake_case__   )
												_A								= self.conv_out(snake_case__   )

												return sample





class 						SCREAMING_SNAKE_CASE				( nn.Module					):





							"""simple docstring"""

							def __init__(  self:   Optional[Any]		,     __A:   int		,     __A:   str		,     __A:   int		,     __A:   Tuple=None		,     __A:   Optional[int]="random"		,     __A:   int=False		,     __A:   Dict=True   )    ->				Any:
												super().__init__()
												_A								= n_e
												_A								= vq_embed_dim
												_A								= beta
												_A								= legacy

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

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

												_A								= sane_index_shape

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

							def       __A (  self:   List[str]		,     __A:   List[str]   )    ->				Optional[int]:
												_A								= inds.shape
												assert len(snake_case__   ) > 1
												_A								= inds.reshape(ishape[0]		,     -1   )
												_A								= self.used.to(snake_case__   )
												if self.re_embed > self.used.shape[0]:  # extra token
																	_A								= 0  # simply set to zero
												_A								= torch.gather(used[None, :][inds.shape[0] * [0], :]		,     1		,     snake_case__   )
												return back.reshape(snake_case__   )

							def       __A (  self:   List[str]		,     __A:   int   )    ->				Any:
												_A								= z.permute(0		,     2		,     3		,     1   ).contiguous()
												_A								= z.view(-1		,     self.vq_embed_dim   )

												# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
												_A								= torch.argmin(torch.cdist(snake_case__		,     self.embedding.weight   )		,     dim=1   )

												_A								= self.embedding(snake_case__   ).view(z.shape   )
												_A								= None
												_A								= None

												# compute loss for embedding
												if not self.legacy:
																	_A								= self.beta * torch.mean((z_q.detach() - z) ** 2   ) + torch.mean((z_q - z.detach()) ** 2   )
												else:
																	_A								= torch.mean((z_q.detach() - z) ** 2   ) + self.beta * torch.mean((z_q - z.detach()) ** 2   )

												# preserve gradients
												_A								= z + (z_q - z).detach()

												# reshape back to match original input shape
												_A								= z_q.permute(0		,     3		,     1		,     2   ).contiguous()

												if self.remap is not None:
																	_A								= min_encoding_indices.reshape(z.shape[0]		,     -1   )  # add batch axis
																	_A								= self.remap_to_used(snake_case__   )
																	_A								= min_encoding_indices.reshape(-1		,     1   )  # flatten

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

												# get quantized latent vectors
												_A								= self.embedding(snake_case__   )

												if shape is not None:
																	_A								= z_q.view(snake_case__   )
																	# reshape back to match original input shape
																	_A								= z_q.permute(0		,     3		,     1		,     2   ).contiguous()

												return z_q





class 						SCREAMING_SNAKE_CASE				( SCREAMING_SNAKE_CASE__					):





							"""simple docstring"""

							def __init__(  self:   Optional[Any]		,     __A:   Optional[Any]		,     __A:   Tuple=False   )    ->				Any:
												_A								= parameters
												_A								= torch.chunk(snake_case__		,     2		,     dim=1   )
												_A								= torch.clamp(self.logvar		,     -30.0		,     20.0   )
												_A								= deterministic
												_A								= torch.exp(0.5 * self.logvar   )
												_A								= torch.exp(self.logvar   )
												if self.deterministic:
																	_A								= torch.zeros_like(
																	    self.mean		,     device=self.parameters.device		,     dtype=self.parameters.dtype   )

							def       __A (  self:   Union[str, Any]		,     __A:   Optional[Any] = None   )    ->				Optional[Any]:
												_A								= randn_tensor(
												    self.mean.shape		,     generator=snake_case__		,     device=self.parameters.device		,     dtype=self.parameters.dtype   )
												_A								= self.mean + self.std * sample
												return x

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



							def       __A (  self:   List[Any]   )    ->				str:
												return self.mean











import math
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils import SchedulerMixin, SchedulerOutput





class 						SCREAMING_SNAKE_CASE				( snake_case  ,	snake_case					):





							"""simple docstring"""

							A_						  =					1

							@register_to_config
							def __init__(  self:   Any		,     __A:   int = 10_00		,     __A:   Optional[Union[np.ndarray, List[float]]] = None   )    ->				List[str]:
												# set `betas`, `alphas`, `timesteps`
												self.set_timesteps(__A   )

												# standard deviation of the initial noise distribution
												_A								= 1.0

												# For now we only support F-PNDM, i.e. the runge-kutta method
												# For more information on the algorithm please take a look at the paper: https://arxiv.org/pdf/2202.09778.pdf
												# mainly at formula (9), (12), (13) and the Algorithm 2.
												_A								= 4

												# running values
												_A								= []

							def       __A (  self:   str		,     __A:   int		,     __A:   Union[str, torch.device] = None   )    ->				int:
												_A								= num_inference_steps
												_A								= torch.linspace(1		,     0		,     num_inference_steps + 1   )[:-1]
												_A								= torch.cat([steps, torch.tensor([0.0]   )]   )

												if self.config.trained_betas is not None:
																	_A								= torch.tensor(self.config.trained_betas		,     dtype=torch.floataa   )
												else:
																	_A								= torch.sin(steps * math.pi / 2   ) ** 2

												_A								= (1.0 - self.betas**2) ** 0.5

												_A								= (torch.atana(self.betas		,     self.alphas   ) / math.pi * 2)[:-1]
												_A								= timesteps.to(__A   )

												_A								= []

							def       __A (  self:   Tuple		,     __A:   torch.FloatTensor		,     __A:   int		,     __A:   torch.FloatTensor		,     __A:   bool = True		,     )    ->				Union[SchedulerOutput, Tuple]:
												if self.num_inference_steps is None:
																	raise ValueError(
																	    '''Number of inference steps is \'None\', you need to run \'set_timesteps\' after creating the scheduler'''   )

												_A								= (self.timesteps == timestep).nonzero().item()
												_A								= timestep_index + 1

												_A								= sample * self.betas[timestep_index] + model_output * self.alphas[timestep_index]
												self.ets.append(__A   )

												if len(self.ets   ) == 1:
																	_A								= self.ets[-1]
												elif len(self.ets   ) == 2:
																	_A								= (3 * self.ets[-1] - self.ets[-2]) / 2
												elif len(self.ets   ) == 3:
																	_A								= (23 * self.ets[-1] - 16 * self.ets[-2] + 5 * self.ets[-3]) / 12
												else:
																	_A								= (1 / 24) * (55 * self.ets[-1] - 59 * self.ets[-2] + 37 * self.ets[-3] - 9 * self.ets[-4])

												_A								= self._get_prev_sample(__A		,     __A		,     __A		,     __A   )

												if not return_dict:
																	return (prev_sample,)

												return SchedulerOutput(prev_sample=__A   )

							def       __A (  self:   Optional[int]		,     __A:   torch.FloatTensor		,     *__A:   Tuple		,     **__A:   List[Any]   )    ->				torch.FloatTensor:
												return sample

							def       __A (  self:   List[str]		,     __A:   Optional[Any]		,     __A:   Optional[Any]		,     __A:   Any		,     __A:   List[Any]   )    ->				List[Any]:
												_A								= self.alphas[timestep_index]
												_A								= self.betas[timestep_index]

												_A								= self.alphas[prev_timestep_index]
												_A								= self.betas[prev_timestep_index]

												_A								= (sample - sigma * ets) / max(__A		,     1e-8   )
												_A								= next_alpha * pred + ets * next_sigma

												return prev_sample



							def __len__(  self:   List[str]   )    ->				Dict:
												return self.config.num_train_timesteps



